<commit_before>package com.dmitriid.tetrad.adapters;

import com.dmitriid.tetrad.interfaces.ITetradCallback;
import com.dmitriid.tetrad.services.FirehoseMessage;
import com.dmitriid.tetrad.utils.JIDUtils;
import com.fasterxml.jackson.databind.JsonNode;
import org.jivesoftware.smack.SmackConfiguration;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.XMPPException;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smackx.muc.DiscussionHistory;
import org.jivesoftware.smackx.muc.MultiUserChat;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class TetradXMPP {
    private final String service_domain;
    private       String chat_service;
    private final String username;
    private final String password;
    private final String resource;
    private Boolean resource_per_user = false;
    private final Integer max_resources;

    private final List<String> ignore = new ArrayList<>();

    private final List<String>               rooms          = new ArrayList<>();
    private final Map<String, MultiUserChat> connectedRooms = new HashMap<>();

    private final HashMap<String/*user*/, MultiUserChat>
            perUserRooms = new HashMap<>();
    private final HashMap<String/*user*/, XMPPConnection>
            perUserConnections = new HashMap<>();

    private XMPPConnection xmppConnection;

    private ITetradCallback callback;

    public TetradXMPP(JsonNode config){
        service_domain = config.at("/service_domain").asText();
        setChatService(config.at("/chat_service").asText());
        username = config.at("/username").asText();
        password = config.at("/password").asText();
        resource = config.at("/resource").asText();
        resource_per_user = config.at("/resource_per_user").asBoolean(false);
        max_resources = config.at("/max_resources").asInt(5);

        for (JsonNode user : config.at("/ignore")) {
            ignore.add(user.asText());
        }

        for (JsonNode room : config.at("/rooms")) {
            rooms.add(room.asText());
        }
    }

    public void start(ITetradCallback callback){
        this.callback = callback;
        try {
            xmppConnection = new XMPPConnection(service_domain);
            xmppConnection.connect();

            xmppConnection.login(username, password, resource);

            DiscussionHistory history = new DiscussionHistory();
            history.setMaxStanzas(0);

            for (String room : rooms) {
                final String roomJID = room + "@" + getChatService();
                MultiUserChat chat = new MultiUserChat(xmppConnection, roomJID);
                chat.join(xmppConnection.getUser(), password, history, SmackConfiguration.getPacketReplyTimeout());
                chat.addMessageListener(packet -> this.handleMessage((Message) packet, roomJID));
                chat.changeNickname("β");
                connectedRooms.put(room, chat);
            }
        } catch (XMPPException e) {
            e.printStackTrace();
        }
    }

    private void handleMessage(Message message, String room) {
        String jid = message.getFrom().replace(room + "/", "");

        long count = ignore.stream()
                           .filter(s -> jid.equals(s) || jid.matches(s))
                           .count();

        if(count > 0){
            return;
        }

        System.out.println(message.getFrom());

        FirehoseMessage firehoseMessage = new FirehoseMessage("xmpp",
                                                              message.getType().name(),
                                                              JIDUtils.bareJID(jid) + "@" + getChatService(),
                                                              getChatService(),
                                                              room,
                                                              message.getBody());
        callback.execute(firehoseMessage);
    }

    public String getChatService() {
        return chat_service;
    }

    private void setChatService(String chat_service) {
        this.chat_service = chat_service;
    }

    public void post(FirehoseMessage firehoseMessage){
        if (!chat_service.equals(firehoseMessage.service)) {
            return;
        }

        if (!connectedRooms.containsKey(firehoseMessage.channel)) {
            return;
        }
        MultiUserChat chat = connectedRooms.get(firehoseMessage.channel);
        if (!chat.isJoined()) return;

        try {
            if(resource_per_user){
                postAsUser(chat, firehoseMessage);
            } else {
                chat.sendMessage(firehoseMessage.user + ": " + firehoseMessage.content);
            }
        } catch (XMPPException e) {
            e.printStackTrace();
        }
    }

    private void postAsUser(MultiUserChat defaultChat, FirehoseMessage firehoseMessage){
        try{
            XMPPConnection perUserConn = new XMPPConnection(service_domain);
            if(!perUserConnections.containsKey(firehoseMessage.user)) {
                if (perUserConnections.size() == max_resources) {
                    defaultChat.sendMessage(firehoseMessage.user + ": " + firehoseMessage.content);
                    return;
                } else {
                    perUserConn.connect();
                    perUserConn.login(username, password, JIDUtils.bareUsername(firehoseMessage.user));
                    perUserConnections.put(firehoseMessage.user, perUserConn);
                }
            }
            MultiUserChat perUserChatRoom;
            String localUser = firehoseMessage.channel + "@" + firehoseMessage.service + "/" + firehoseMessage.user;
            if(perUserRooms.containsKey(localUser)){
                perUserChatRoom = perUserRooms.get(localUser);
            } else {
                DiscussionHistory history = new DiscussionHistory();
                history.setMaxStanzas(0);

                final String roomJID = firehoseMessage.channel + "@" + getChatService();
                perUserChatRoom = new MultiUserChat(perUserConn, roomJID);<commit_msg>Use "strong" for usernames in xxx -> XMPP<commit_after>package com.dmitriid.tetrad.adapters;

import com.dmitriid.tetrad.interfaces.ITetradCallback;
import com.dmitriid.tetrad.services.FirehoseMessage;
import com.dmitriid.tetrad.utils.JIDUtils;
import com.fasterxml.jackson.databind.JsonNode;
import org.jivesoftware.smack.SmackConfiguration;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.XMPPException;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smackx.XHTMLManager;
import org.jivesoftware.smackx.XHTMLText;
import org.jivesoftware.smackx.muc.DiscussionHistory;
import org.jivesoftware.smackx.muc.MultiUserChat;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class TetradXMPP {
    private final String service_domain;
    private       String chat_service;
    private final String username;
    private final String password;
    private final String resource;
    private Boolean resource_per_user = false;
    private final Integer max_resources;

    private final List<String> ignore = new ArrayList<>();

    private final List<String>               rooms          = new ArrayList<>();
    private final Map<String, MultiUserChat> connectedRooms = new HashMap<>();

    private final HashMap<String/*user*/, MultiUserChat>
            perUserRooms = new HashMap<>();
    private final HashMap<String/*user*/, XMPPConnection>
            perUserConnections = new HashMap<>();

    private XMPPConnection xmppConnection;

    private ITetradCallback callback;

    public TetradXMPP(JsonNode config){
        service_domain = config.at("/service_domain").asText();
        setChatService(config.at("/chat_service").asText());
        username = config.at("/username").asText();
        password = config.at("/password").asText();
        resource = config.at("/resource").asText();
        resource_per_user = config.at("/resource_per_user").asBoolean(false);
        max_resources = config.at("/max_resources").asInt(5);

        for (JsonNode user : config.at("/ignore")) {
            ignore.add(user.asText());
        }

        for (JsonNode room : config.at("/rooms")) {
            rooms.add(room.asText());
        }
    }

    public void start(ITetradCallback callback){
        this.callback = callback;
        try {
            xmppConnection = new XMPPConnection(service_domain);
            xmppConnection.connect();

            xmppConnection.login(username, password, resource);

            DiscussionHistory history = new DiscussionHistory();
            history.setMaxStanzas(0);

            for (String room : rooms) {
                final String roomJID = room + "@" + getChatService();
                MultiUserChat chat = new MultiUserChat(xmppConnection, roomJID);
                chat.join(xmppConnection.getUser(), password, history, SmackConfiguration.getPacketReplyTimeout());
                chat.addMessageListener(packet -> this.handleMessage((Message) packet, roomJID));
                chat.changeNickname("β");
                connectedRooms.put(room, chat);
            }
        } catch (XMPPException e) {
            e.printStackTrace();
        }
    }

    private void handleMessage(Message message, String room) {
        String jid = message.getFrom().replace(room + "/", "");

        long count = ignore.stream()
                           .filter(s -> jid.equals(s) || jid.matches(s))
                           .count();

        if(count > 0){
            return;
        }

        System.out.println(message.getFrom());

        FirehoseMessage firehoseMessage = new FirehoseMessage("xmpp",
                                                              message.getType().name(),
                                                              JIDUtils.bareJID(jid) + "@" + getChatService(),
                                                              getChatService(),
                                                              room,
                                                              message.getBody());
        callback.execute(firehoseMessage);
    }

    public String getChatService() {
        return chat_service;
    }

    private void setChatService(String chat_service) {
        this.chat_service = chat_service;
    }

    public void post(FirehoseMessage firehoseMessage){
        if (!chat_service.equals(firehoseMessage.service)) {
            return;
        }

        if (!connectedRooms.containsKey(firehoseMessage.channel)) {
            return;
        }
        MultiUserChat chat = connectedRooms.get(firehoseMessage.channel);
        if (!chat.isJoined()) return;

        if(resource_per_user){
            postAsUser(chat, firehoseMessage);
        } else {
            postDefaultMessage(chat, firehoseMessage);
        }
    }

    private void postDefaultMessage(MultiUserChat chat, FirehoseMessage firehoseMessage){
        Message msg = chat.createMessage();
        msg.setBody(firehoseMessage.user + ": " + firehoseMessage.content);

        XHTMLText xhtml = new XHTMLText(null, null);
        xhtml.appendOpenStrongTag();
        xhtml.append(firehoseMessage.user);
        xhtml.appendCloseStrongTag();
        xhtml.append(": ");
        xhtml.append(firehoseMessage.content);

        XHTMLManager.addBody(msg, xhtml.toString());
        try {
            chat.sendMessage(msg);
        } catch (XMPPException e) {
            try {
                chat.sendMessage(firehoseMessage.user + ": " + firehoseMessage.content);
            } catch (XMPPException e1) {
                e1.printStackTrace();
            }
            e.printStackTrace();
        }
    }

    private void postAsUser(MultiUserChat defaultChat, FirehoseMessage firehoseMessage){
        try{
            XMPPConnection perUserConn = new XMPPConnection(service_domain);
            if(!perUserConnections.containsKey(firehoseMessage.user)) {
                if (perUserConnections.size() == max_resources) {
                    postDefaultMessage(defaultChat, firehoseMessage);
                    return;
                } else {
                    perUserConn.connect();
                    perUserConn.login(username, password, JIDUtils.bareUsername(firehoseMessage.user));
                    perUserConnections.put(firehoseMessage.user, perUserConn);
                }
            }
            MultiUserChat perUserChatRoom;
            String localUser = firehoseMessage.channel + "@" + firehoseMessage.service + "/" + firehoseMessage.user;
            if(perUserRooms.containsKey(localUser)){
                perUserChatRoom = perUserRooms.get(localUser);
            } else {
                DiscussionHistory history = new DiscussionHistory();
                history.setMaxStanzas(0);

                final String roomJID = firehoseMessage.channel + "@" + getChatService();
                perUserChatRoom = new MultiUserChat(perUserConn, roomJID);
{'fraction_non_alphanumeric': 0.1011637635962577, 'fraction_numerical': 0.0007606298014756218, 'mean_word_length': 2.1149016820658613, 'pattern_counts': {'":': 5, '<': 23, '<?xml version=': 0, '>': 30, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 21, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2889512', 'n_tokens_mistral': 3508, 'n_tokens_neox': 3367, 'n_words': 624}