feat: stream event handler

2 files changed, 109 insertions, 6 deletions

diff --git a/src/stream.cpp b/src/stream.cpp
index a42a050..2fa1a34 100644
--- a/src/stream.cpp
+++ b/src/stream.cpp
@@ -2,7 +2,17 @@
 
 #include <algorithm>
 #include <chrono>
+#include <pqxx/pqxx>
+#include <set>
+#include <string>
 #include <thread>
+#include <utility>
+#include <vector>
+
+#include "api/twitch/schemas/stream.hpp"
+#include "config.hpp"
+#include "schemas/stream.hpp"
+#include "utils/string.hpp"
 
 namespace bot::stream {
   void StreamListenerClient::listen_channel(const int &id) {
@@ -44,18 +54,109 @@ namespace bot::stream {
 
       if (!is_already_live) {
         this->online_ids.insert(stream.user_id);
+        this->handler(schemas::EventType::LIVE, stream);
       }
     }
 
     // removing old ids
-    for (auto i = this->online_ids.begin(); i != this->online_ids.end(); i++) {
-      bool is_live =
-          std::any_of(streams.begin(), streams.end(),
-                      [&](const auto &x) { return x.user_id == *i; });
+    for (auto i = this->online_ids.begin(); i != this->online_ids.end();) {
+      auto stream =
+          std::find_if(streams.begin(), streams.end(),
+                       [&](const auto &x) { return x.user_id == *i; });
+
+      if (stream == streams.end()) {
+        this->handler(schemas::EventType::OFFLINE,
+                      api::twitch::schemas::Stream{*i});
+        i = this->online_ids.erase(i);
+      } else {
+        ++i;
+      }
+    }
+  }
+  void StreamListenerClient::handler(
+      const schemas::EventType &type,
+      const api::twitch::schemas::Stream &stream) {
+    pqxx::connection conn(GET_DATABASE_CONNECTION_URL(this->configuration));
+    pqxx::work work(conn);
+
+    pqxx::result events = work.exec(
+        "SELECT id, channel_id, message, flags FROM events WHERE event_type "
+        "= " +
+        std::to_string(type) +
+        " AND target_alias_id = " + std::to_string(stream.user_id));
+
+    for (const auto &event : events) {
+      pqxx::row channel =
+          work.exec1("SELECT alias_id, alias_name FROM channels WHERE id = " +
+                     std::to_string(event[1].as<int>()));
+
+      pqxx::result subs = work.exec(
+          "SELECT user_id FROM event_subscriptions WHERE event_id = " +
+          std::to_string(event[0].as<int>()));
+
+      std::set<std::string> user_ids;
+      if (!subs.empty()) {
+        for (const auto &sub : subs) {
+          user_ids.insert(std::to_string(sub[0].as<int>()));
+        }
+
+        pqxx::result users = work.exec(
+            "SELECT alias_name FROM users WHERE id IN (" +
+            utils::string::str(user_ids.begin(), user_ids.end(), ',') + ")");
+
+        user_ids.clear();
+
+        for (const auto &user : users) {
+          user_ids.insert(user[0].as<std::string>());
+        }
+      }
+
+      auto flags = event[3].as_array();
+      std::pair<pqxx::array_parser::juncture, std::string> elem;
+
+      do {
+        elem = flags.get_next();
+        if (elem.first == pqxx::array_parser::juncture::string_value) {
+          if (std::stoi(elem.second) == schemas::EventFlag::MASSPING) {
+            auto chatters = this->helix_client.get_chatters(
+                channel[0].as<int>(), this->irc_client.get_bot_id());
+
+            for (const auto &chatter : chatters) {
+              user_ids.insert(chatter.login);
+            }
+          }
+        }
+      } while (elem.first != pqxx::array_parser::juncture::done);
+
+      std::string base = "⚡ " + event[2].as<std::string>();
+      std::vector<std::string> msgs = {""};
+      int index = 0;
+
+      if (!user_ids.empty()) {
+        base.append(" · ");
+      }
+
+      for (const auto &user_id : user_ids) {
+        const std::string &current_msg = msgs.at(index);
+        std::string x = "@" + user_id;
+
+        if (base.length() + current_msg.length() + 1 + x.length() >= 500) {
+          index += 1;
+        }
 
-      if (!is_live) {
-        this->online_ids.erase(i);
+        if (index > msgs.size() - 1) {
+          msgs.push_back(x);
+        } else {
+          msgs[index] = current_msg + " " + x;
+        }
       }
+
+      for (const auto &msg : msgs) {
+        this->irc_client.say(channel[1].as<std::string>(), base + msg);
+      }
+
+      work.commit();
+      conn.close();
     }
   }
 }
diff --git a/src/stream.hpp b/src/stream.hpp
index 687fc1e..7701cb3 100644
--- a/src/stream.hpp
+++ b/src/stream.hpp
@@ -28,6 +28,8 @@ namespace bot::stream {
     private:
       void run();
       void check();
+      void handler(const schemas::EventType &type,
+                   const api::twitch::schemas::Stream &stream);
 
       const api::twitch::HelixClient &helix_client;
       irc::Client &irc_client;