[MOD/FIX] Separate XMPP thread, fix EINVAL issue

threading is good actually. please hmu if you ever trigger the
achievement.

Makefile

@@ -19,8 +19,8 @@ SOURCE=src
 OBJECT=build
 INCLUDES=src/include
 CC=cc
-CFLAGS=-I$(INCLUDES) -I$(CYTO_INC) -DNAME="\"$(NAME)\"" -DVERSION="\"$(VERSION)\"" -DREPOSITORY=\"$(REPOSITORY)\" -O3 -g -ggdb -Wall -Werror
+CFLAGS=-I$(SOURCE) -I$(INCLUDES) -I$(CYTO_INC) -DNAME="\"$(NAME)\"" -DVERSION="\"$(VERSION)\"" -DREPOSITORY=\"$(REPOSITORY)\" -O2 -g -ggdb -Wall -Werror
-LDFLAGS=-L $(CYTO_LIB) -lCytoplasm -Wl,--export-dynamic -O3 -g -ggdb
+LDFLAGS=-L $(CYTO_LIB) -lCytoplasm -Wl,--export-dynamic -O2 -g -ggdb
 BINARY=parsee
 # ============================ Compilation =================================
 SRC_FILES:=$(shell find $(SOURCE) -name '*.c')

XEPS-TBD.TXT

@@ -55,3 +55,5 @@ Not XEPs, but ideas that _needs_ to be added:
 
 ------------------------- MSC LAND -------------------------
 - MSC2346: Bridged metadata is useful, at least for other bridges.
+- https://spec.matrix.org/v1.11/client-server-api/#third-party-lookups is _very_ 
+useful on the Matrix side, to be frank.

src/Command/Parser.c

@@ -15,7 +15,7 @@ Command *
 CommandParse(char *cmd)
 {
     Command *ret;
-    char *end_data, *cur, *namestart;
+    char *end_data, *cur, *namestart = NULL;
     char *key = NULL, *val = NULL;
     size_t len;
     CommandState state = STATE_WHITE;
@@ -60,7 +60,7 @@ CommandParse(char *cmd)
                 }
                 break;
             case STATE_NAME:
-                if (c == '=')
+                if (c == '=' && namestart)
                 {
                     len = cur - namestart;
                     key = Malloc(len + 1);

src/Main.c

@@ -99,11 +99,6 @@ Main(void)
         goto end;
     }
 
-    Log(LOG_NOTICE, "Listening to MUCs...");
-    LogConfigIndent(LogConfigGlobal());
-    ParseeSendPresence(conf.handlerArgs);
-    LogConfigUnindent(LogConfigGlobal());
-
     server = HttpServerCreate(&conf);
 
     if (!ParseeInitialiseSignals(server, xmpp_thr, jabber))
@@ -112,6 +107,13 @@ Main(void)
     }
 
     HttpServerStart(server);
+
+    Log(LOG_NOTICE, "Listening to MUCs...");
+    LogConfigIndent(LogConfigGlobal());
+    ParseeSendPresence(conf.handlerArgs);
+    LogConfigUnindent(LogConfigGlobal());
+
+
     LogConfigUnindent(LogConfigGlobal());
     Log(LOG_INFO, "=======================");
     HttpServerJoin(server);

src/MatrixEventHandler.c

@@ -56,7 +56,6 @@ ParseeMemberHandler(ParseeData *data, HashMap *event)
             char *muc = ParseeGetMUCID(data, chat_id);
             char *rev = StrConcat(2, muc, "/parsee");
 
-            /* Make the user join the MUC */
             XMPPJoinMUC(data->jabber, jid, rev);
             Free(rev);
             Free(muc);
@@ -295,7 +294,9 @@ ParseeMessageHandler(ParseeData *data, HashMap *event)
         }
         xmppified_user = ParseeEncodeMXID(m_sender);
 
-        /* TODO: Check the name's validity */
+        /* TODO: Check the name's validity.
+         * Is there a good way to check for that that isn't 
+         * just "await on join and try again?" */
         name = ASGetName(data->config, id, m_sender);
         rev = StrConcat(4, muc_id, "/", name, "[p]");
 
@@ -382,7 +383,9 @@ ParseeEventHandler(ParseeData *data, HashMap *event)
         ParseeMemberHandler(data, event);        
         return;
     }
-    else if (StrEquals(event_type, "m.room.message"))
+    else if (StrEquals(event_type, "m.room.message") ||
+             StrEquals(event_type, "m.sticker"))    /* TODO: Actual sticker
+                                                     * support here... */
     {
         ParseeMessageHandler(data, event);        
         return;

src/Parsee/User.c

@@ -744,3 +744,15 @@ ParseeIsAdmin(ParseeData *data, char *user)
     DbUnlock(data->db, admins);
     return ret;
 }
+
+char *
+ParseeJID(ParseeData *data)
+{
+    return StrConcat(2, "parsee@", data->config->component_host);
+}
+char *
+ParseeMXID(ParseeData *data)
+{
+    return StrConcat(4, "@", data->config->sender_localpart,
+                        ":", data->config->homeserver_host);
+}

src/XMPP/Stanza.c

@@ -281,14 +281,15 @@ XMPPLeaveMUC(XMPPComponent *comp, char *fr, char *muc, char *reason)
 
     pthread_mutex_unlock(&comp->write_lock);
 }
-void
+
+bool
 XMPPJoinMUC(XMPPComponent *comp, char *fr, char *muc)
 {
-    XMLElement *presence, *x;
+    XMLElement *presence, *x, *reply;
     char *from, *id;
     if (!comp || !fr || !muc)
     {
-        return;
+        return false;
     }
 
     pthread_mutex_lock(&comp->write_lock);
@@ -308,9 +309,25 @@ XMPPJoinMUC(XMPPComponent *comp, char *fr, char *muc)
 
     XMLFreeElement(presence);
     Free(from);
-    Free(id);
 
     pthread_mutex_unlock(&comp->write_lock);
+
+    if ((reply = ParseeAwaitStanza(id, 500)))
+    {
+        bool exit_code = true;
+
+        if (XMPPHasError(reply, "conflict"))
+        {
+            Log(LOG_WARNING, "UNIMPLEMENTED NAMING CONFLICT.");
+            exit_code = false;
+        }
+
+        XMLFreeElement(reply);
+        Free(id);
+        return exit_code;
+    }
+    Free(id);
+    return true;
 }
 bool
 XMPPIsParseeStanza(XMLElement *stanza)
@@ -432,7 +449,6 @@ XMPPGetModeration(XMLElement *stanza)
     moderate = XMLookForTKV(stanza, "moderate", "xmlns", MOD_BASE ":1");
     if (moderate)
     {
-        Log(LOG_WARNING, "0.3.0 moderation");
         return HashMapGet(moderate->attrs, "id");
     }
     
@@ -452,3 +468,20 @@ XMPPGetModeration(XMLElement *stanza)
     return NULL;
 #undef MOD_BASE
 }
+
+bool 
+XMPPHasError(XMLElement *stanza, char *type)
+{
+    XMLElement *err;
+    if (!stanza || !type)
+    {
+        return false;
+    }
+
+    err = XMLookForUnique(stanza, "error");
+    if (!err)
+    {
+        return false;
+    }
+    return XMLookForUnique(err, type);
+}

src/XMPPThread/Bridged.c

@@ -0,0 +1,113 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+
+char *
+ParseeGetBridgedRoom(ParseeData *data, XMLElement *stanza)
+{
+    char *to = ParseeDecodeMXID(HashMapGet(stanza->attrs, "to"));
+    char *from = HashMapGet(stanza->attrs, "from");
+    char *chat_id = ParseeGetFromMUCID(data, from);
+    char *mroom_id = ParseeGetRoomID(data, chat_id);
+    char *ret;
+
+    Free(chat_id);
+    if (mroom_id)
+    {
+        Free(to);
+        return mroom_id;
+    }
+
+    /* Not a MUC, find the DMd room. */
+    ret = ParseeFindDMRoom(data, to, from);
+    Free(to);
+    return ret;
+}
+
+
+char *
+ParseeGetEventFromID(ParseeData *data, XMLElement *stanza, char *id)
+{
+    char *from = (HashMapGet(stanza->attrs, "from"));
+    char *chat_id = ParseeGetFromMUCID(data, from);
+    char *ret = ParseeEventFromSID(data, chat_id, id);
+    char *mroom_id = ParseeGetBridgedRoom(data, stanza);
+
+    if (!ret)
+    {
+        Free(ret);
+        ret = ParseeEventFromID(data, chat_id, id);
+    }
+
+    Free(chat_id);
+    if (ret)
+    {
+        Free(mroom_id);
+        return ret;
+    }
+    ret = ParseeDMEventFromID(data, mroom_id, id);
+    Free(mroom_id);
+
+    return ret;
+}
+char *
+ParseeGetReactedEvent(ParseeData *data, XMLElement *stanza)
+{
+    XMLElement *reactions = XMLookForTKV(stanza,
+        "reactions", "xmlns", "urn:xmpp:reactions:0"
+    );
+    char *reacted_id = reactions ? HashMapGet(reactions->attrs, "id") : NULL;
+
+    return ParseeGetEventFromID(data, stanza, reacted_id);
+}
+
+
+
+void
+ParseePushAllStanza(ParseeData *args, XMLElement *stanza, char *event)
+{
+    char *xmpp_from = HashMapGet(stanza->attrs, "from");
+    char *mroom_id  = ParseeGetBridgedRoom(args, stanza);
+    char *chat_id   = ParseeGetFromMUCID(args, xmpp_from);
+
+    char *id_str    = HashMapGet(stanza->attrs, "id");
+    char *s_id_str  = XMPPGetStanzaID(stanza);
+
+    if (!chat_id)
+    {
+        ParseePushDMStanza(
+            args, mroom_id, s_id_str, 
+            id_str, event, xmpp_from
+        );
+    }
+    ParseePushStanza(args, chat_id, s_id_str, id_str, event, xmpp_from);
+    Free(mroom_id);
+    Free(chat_id);
+}
+
+bool
+ParseeVerifyAllStanza(ParseeData *args, XMLElement *stanza)
+{
+    char *to, *room;
+    char *from    = HashMapGet(stanza->attrs, "from");
+    char *id      = HashMapGet(stanza->attrs, "id");
+    char *chat_id = ParseeGetFromMUCID(args, from);
+    bool ret;
+
+    if (chat_id)
+    {
+        char *s_id_str  = XMPPGetStanzaID(stanza);
+        ret = ParseeVerifyStanza(args, chat_id, s_id_str);
+
+        Free(chat_id);
+        return ret;
+    }
+    to = ParseeDecodeMXID(HashMapGet(stanza->attrs, "to"));
+    room = ParseeFindDMRoom(args, to, from);
+    ret = ParseeVerifyDMStanza(args, chat_id, id);
+
+    Free(room);
+    Free(to);
+
+    return ret;
+}

src/XMPPThread/ICollate.c

@@ -0,0 +1,54 @@
+#include <stdbool.h>
+#include <string.h>
+
+#include "XMPPThread/internal.h"
+
+/* Performs i;octet collation. */
+int
+ICollate(unsigned char *cata, unsigned char *catb)
+{
+    size_t al = cata ? strlen((char *) cata) : 0;
+    size_t bl = catb ? strlen((char *) catb) : 0;
+
+    if (!al && !bl)
+    {
+        return 0;
+    }
+    while (true)
+    {
+        if (!al && bl)
+        {
+            return -1;
+        }
+        else if (al && !bl)
+        {
+            return 1;
+        }
+
+        if (*cata == *catb)
+        {
+            cata++;
+            catb++;
+            al--;
+            bl--;
+            continue;
+        }
+        else if (*cata < *catb)
+        {
+            return -1;
+        }
+        return 1;
+    }
+    return 0;
+}
+
+int 
+IdentitySort(void *idap, void *idbp)
+{
+    XMPPIdentity *ida = idap;
+    XMPPIdentity *idb = idbp;
+    unsigned char *cata = (unsigned char *) ida->category;
+    unsigned char *catb = (unsigned char *) idb->category;
+
+    return ICollate(cata, catb);
+}

src/XMPPThread/Pubsub.c

@@ -0,0 +1,31 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Str.h>
+
+#include <XMPP.h>
+#include <XML.h>
+
+XMLElement *
+CreatePubsubRequest(char *from, char *to, char *node)
+{
+    XMLElement *iq_req, *pubsub, *sub;
+    char *id;
+    iq_req = XMLCreateTag("iq");
+    XMLAddAttr(iq_req, "from", from);
+    XMLAddAttr(iq_req, "to", to);
+    XMLAddAttr(iq_req, "id", (id = StrRandom(16)));
+    XMLAddAttr(iq_req, "type", "set");
+
+    pubsub = XMLCreateTag("pubsub");
+    XMLAddAttr(pubsub, "xmlns", "http://jabber.org/protocol/pubsub");
+    XMLAddChild(iq_req, pubsub);
+
+    sub = XMLCreateTag("subscribe");
+    XMLAddAttr(sub, "node", node);
+    XMLAddAttr(sub, "jid", from);
+    XMLAddChild(pubsub, sub);
+
+    Free(id);
+    return iq_req;
+}

src/XMPPThread/PubsubAvatar.c

@@ -0,0 +1,85 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Json.h>
+#include <Cytoplasm/Str.h>
+
+#include <Parsee.h>
+#include <XMPP.h>
+#include <XML.h>
+
+/* Manages an avatar metadata pubsub item */
+static XMLElement *
+CreateAvatarRequest(char *from, char *to, char *avatar_id)
+{
+    XMLElement *iq_req, *pubsub, *items, *item;
+    char *id;
+    iq_req = XMLCreateTag("iq");
+    XMLAddAttr(iq_req, "from", from);
+    XMLAddAttr(iq_req, "to", to);
+    XMLAddAttr(iq_req, "id", (id = StrRandom(16)));
+    XMLAddAttr(iq_req, "type", "get");
+
+    pubsub = XMLCreateTag("pubsub");
+    XMLAddAttr(pubsub, "xmlns", "http://jabber.org/protocol/pubsub");
+    XMLAddChild(iq_req, pubsub);
+
+    items = XMLCreateTag("items");
+    XMLAddAttr(items, "node", "urn:xmpp:avatar:data");
+    XMLAddChild(pubsub, items);
+
+    item = XMLCreateTag("item");
+    XMLAddAttr(item, "id", avatar_id);
+    XMLAddChild(items, item);
+
+    Free(id);
+    return iq_req;
+}
+
+
+void
+ManageProfileItem(ParseeData *args, XMLElement *item, XMLElement *stanza, XMPPThread *thr)
+{
+    XMPPComponent *jabber = args->jabber;
+    DbRef *avatars;
+    HashMap *json;
+
+    char *publisher = HashMapGet(item->attrs, "publisher");
+    char *id = HashMapGet(item->attrs, "id");
+    char *mxid;
+
+    avatars = DbLock(args->db, 1, "avatars");
+    if (!avatars)
+    {
+        avatars = DbCreate(args->db, 1, "avatars");
+    }
+    json = DbJson(avatars);
+
+    mxid = GrabString(json, 1, publisher);
+    if (mxid && StrEquals(mxid, id))
+    {
+        /* We have nothing to do. */
+        goto end;
+    }
+
+    /* We need to download the media to push it. Let's submit a pubsub request. */
+    {
+        char *from = HashMapGet(stanza->attrs, "to");
+        char *to = HashMapGet(stanza->attrs, "from");
+        char *url = HashMapGet(item->attrs, "url");
+        XMLElement *request = CreateAvatarRequest(from, to, id);
+
+        pthread_mutex_lock(&jabber->write_lock);
+        XMLEncode(jabber->stream, request);
+        StreamFlush(jabber->stream);
+        pthread_mutex_unlock(&jabber->write_lock);
+
+        XMLFreeElement(request);
+
+        (void) url; /* TODO */
+    }
+
+end:
+    DbUnlock(args->db, avatars);
+}
+

src/XMPPThread/ReListener.c

@@ -0,0 +1,382 @@
+/* The reader-listeners system for the XPPP thread. */
+#include <Parsee.h>
+
+#include <pthread.h>
+#include <string.h>
+#include <ctype.h>
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Base64.h>
+#include <Cytoplasm/Util.h>
+#include <Cytoplasm/Log.h>
+#include <Cytoplasm/Str.h>
+#include <Cytoplasm/Sha.h>
+
+#include <StringStream.h>
+#include <XMPPCommand.h>
+#include <Matrix.h>
+#include <XMPP.h>
+#include <XML.h>
+#include <AS.h>
+
+#include "XMPPThread/internal.h"
+
+/* Generates a SHA-256 hash of the ver field. */
+char *
+XMPPGenerateVer(void)
+{
+    char *S = NULL;
+    unsigned char *Sha = NULL;
+    Array *identities = ArrayCreate();
+    Array *features = ArrayCreate();
+    size_t i;
+
+    /* Initialise identity table, to be sorted */
+#define IdentitySimple(cat, Type, Name) { \
+    XMPPIdentity *id = Malloc(sizeof(*id)); \
+    id->category = cat; \
+    id->lang = NULL; \
+    id->type = Type; \
+    id->name = Name; \
+    ArrayAdd(identities, id); }
+    IQ_IDENTITY
+#undef IdentitySimple
+#define AdvertiseSimple(feature) ArrayAdd(features, feature);
+    IQ_ADVERT
+#undef AdvertiseSimple
+    ArraySort(identities, IdentitySort);
+    for (i = 0; i < ArraySize(identities); i++)
+    {
+        XMPPIdentity *identity = ArrayGet(identities, i);
+        char *id_chunk = StrConcat(7, 
+            identity->category, "/",
+            identity->type, "/",
+            identity->lang, "/",
+            identity->name);
+        char *tmp = S;
+        S = StrConcat(3, S, id_chunk, "<");
+        Free(tmp);
+        Free(id_chunk);
+    }
+
+    ArraySort(features, ((int (*) (void *, void *)) ICollate));
+    for (i = 0; i < ArraySize(features); i++)
+    {
+        char *feature = ArrayGet(features, i);
+        char *tmp = S;
+        S = StrConcat(3, S, feature, "<");
+        Free(tmp);
+    }
+
+    Sha = Sha1(S);
+    Free(S);
+    S = Base64Encode((const char *) Sha, 20);
+    Free(Sha);
+
+    ArrayFree(features);
+    for (i = 0; i < ArraySize(identities); i++)
+    {
+        XMPPIdentity *identity = ArrayGet(identities, i);
+        /* We don't have to do anything here. */
+        Free(identity);
+    }
+    ArrayFree(identities);
+
+    return S;
+}
+
+XMLElement *
+RetrieveStanza(XMPPThread *thread)
+{
+    XMLElement *ret = NULL;
+
+    pthread_mutex_lock(&thread->info->lock);
+    ret = ArrayDelete(thread->info->stanzas, 0);
+    pthread_mutex_unlock(&thread->info->lock);
+
+    return ret;
+}
+static void
+PushStanza(XMPPThreadInfo *info, XMLElement *stanza)
+{
+    pthread_mutex_lock(&info->lock);
+    ArrayAdd(info->stanzas, stanza);
+    pthread_mutex_unlock(&info->lock);
+}
+
+static void *
+XMPPDispatcher(void *argp)
+{
+    XMPPThread *thread = argp;
+    ParseeData *args = thread->info->args;
+
+    while (thread->info->running)
+    {
+        XMLElement *stanza = RetrieveStanza(thread);
+ 
+        /* TODO: I've seen some spikes in some threads. */
+        if (!stanza)
+        {
+            UtilSleepMillis(5);
+            continue;
+        }
+
+        if (StrEquals(stanza->name, "presence"))
+        {
+            PresenceStanza(args, stanza);
+            XMLFreeElement(stanza);
+            continue;
+        }
+        else if (StrEquals(stanza->name, "message"))
+        {
+            if (!MessageStanza(args, stanza, thread))
+            {
+                continue;
+            }
+        }
+        else if (StrEquals(stanza->name, "iq"))
+        {
+            IQStanza(args, stanza, thread);
+        }
+        else
+        {
+            Log(LOG_WARNING, "Unknown stanza '%s':", stanza->name);
+            XMLEncode(StreamStdout(), stanza);
+            StreamPrintf(StreamStdout(), "\n");
+            StreamFlush(StreamStdout());
+        }
+        XMLFreeElement(stanza);
+    }
+
+    return NULL;
+}
+
+typedef struct XMPPAwait {
+    pthread_mutex_t cond_lock;
+    pthread_cond_t condition;
+    bool usable;
+
+    XMLElement *stanza;
+} XMPPAwait;
+
+static pthread_mutex_t await_lock = PTHREAD_MUTEX_INITIALIZER;
+static HashMap *await_table = NULL;
+static XMPPThreadInfo info;
+
+size_t
+ParseeCongestion(void)
+{
+    size_t congestion;
+    pthread_mutex_lock(&info.lock);
+    congestion = ArraySize(info.stanzas);
+    pthread_mutex_unlock(&info.lock);
+
+    return congestion;
+}
+
+void *
+ParseeXMPPThread(void *argp)
+{
+    ParseeData *args = argp;
+    XMPPComponent *jabber = args->jabber;
+    XMLElement *stanza = NULL;
+    size_t i;
+        
+    /* Initialise the await table */
+    await_table = HashMapCreate();
+
+    /* Initialise the command manager, and add all ad-hoc commands */
+    info.m = XMPPCreateManager(args);
+    {
+        XMPPCommand *cmd;
+#define XMPP_COMMAND(f,n,t,s) \
+        cmd =  XMPPBasicCmd( \
+            n, t, f \
+        ); \
+        s \
+        XMPPRegisterCommand(info.m, cmd);
+        XMPPCOMMANDS
+#undef XMPP_COMMAND
+    }
+
+    /* Initialise the FIFO */
+    info.stanzas = ArrayCreate();
+    pthread_mutex_init(&info.lock, NULL);
+
+    /* ... and its readers. */
+    /* TODO: Make that configurable. */
+    info.available_dispatchers = 16;
+    info.dispatchers = Malloc(
+        sizeof(*info.dispatchers) * info.available_dispatchers
+    );
+
+    info.args = args;
+    info.jabber = jabber;
+    info.running = true;
+    pthread_mutex_init(&info.chk_lock, NULL);
+
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t *thr = &info.dispatchers[i].thr;
+        info.dispatchers[i].info = &info;
+
+        pthread_create(thr, NULL, XMPPDispatcher, &info.dispatchers[i]);
+    }
+
+    while (true)
+    {
+        char *id;
+
+        stanza = XMLDecode(jabber->stream, false);
+        if (!stanza)
+        {
+            break;
+        }
+
+        id = HashMapGet(stanza->attrs, "id");
+        if (id)
+        {
+            XMPPAwait *await;
+            /* Lock out the table to see if we're awaiting. */
+            pthread_mutex_lock(&await_lock);
+            if ((await = HashMapGet(await_table, id)))
+            {
+                pthread_mutex_lock(&await->cond_lock);
+                await->stanza = stanza;             
+                pthread_cond_broadcast(&await->condition);
+                pthread_mutex_unlock(&await->cond_lock);
+
+                HashMapDelete(await_table, id);
+
+                pthread_mutex_unlock(&await_lock);
+                continue;
+            }
+            pthread_mutex_unlock(&await_lock);
+        }
+
+        /* Push it into the stanza FIFO. A dispatcher thread should then 
+         * be able to freely grab a value(locked by a mutex). We can't make 
+         * dispatchers read stanzas on their own, since that's _not_ how 
+         * streams work, but this should mitigate some issues, and allow a 
+         * few threads to be busy, while the rest of Parsee works. */
+        PushStanza(&info, stanza);
+    }
+
+    info.running = false;
+    Log(LOG_INFO, "Joining subthreads...");
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t thr = info.dispatchers[i].thr;
+        pthread_join(thr, NULL);
+    }
+    Log(LOG_INFO, "Joined subthreads...");
+    Free(info.dispatchers);
+
+    for (i = 0; i < ArraySize(info.stanzas); i++)
+    {
+        XMLFreeElement(ArrayGet(info.stanzas, i));
+    }
+    ArrayFree(info.stanzas);
+
+    HashMapFree(await_table);
+
+    pthread_mutex_destroy(&info.lock);
+
+    XMPPFreeManager(info.m);
+    return NULL;
+}
+
+#include <time.h>
+#include <errno.h>
+
+/* TODO: This function does NIT behave well. */
+XMLElement *
+ParseeAwaitStanza(char *identifier, int64_t timeout)
+{
+    /* TODO: Pthreads HATE me using Malloc here, so I'm abusing stackspace.
+     * Not *too much* of a problem, just a weird oddity. If anyone has a clue 
+     * on why that happens (at least on ARM64 with a Pi4 running Debian), let 
+     * me know! */
+    XMPPAwait awa;
+    XMLElement *stanza;
+    struct timespec ts;
+    if (!identifier)
+    {
+        return NULL;
+    }
+
+    /* Convert into an absolute timeout.
+     * =================================
+     * XXX: For anyone using timespecs: MAKE ABSOLUTELY FUCKING SURE YOUR NANOS 
+     * ARE IN RANGE. THIS UNIRONICALLY CAUSED SOME REAL CONCURRENCY ERRORS ON MY 
+     * SIDE FROM TV_NSEC OVERFLOWING AND TIMEDWAIT RETURNING EINVAL. */
+    if (timeout > 0)
+    {
+        int64_t seconds = timeout / (1 SECONDS);
+        int64_t msecond = timeout % (1 SECONDS);
+
+        clock_gettime(CLOCK_REALTIME, &ts);
+
+        ts.tv_sec += seconds            ;
+        ts.tv_nsec+= msecond * 1000000  ;
+        if (ts.tv_nsec > 999999999)
+        {
+            int64_t sec_delta = ts.tv_nsec / 1000000000;
+            int64_t nsc_delta = ts.tv_nsec % 1000000000;
+
+            ts.tv_nsec = nsc_delta;
+            ts.tv_sec += sec_delta;
+        }
+
+        (void) msecond;
+    }
+
+    pthread_mutex_lock(&await_lock);
+
+    pthread_cond_init(&awa.condition, NULL);
+    pthread_mutex_init(&awa.cond_lock, NULL);
+    awa.stanza = NULL;
+    
+    HashMapSet(await_table, identifier, &awa);
+    pthread_mutex_unlock(&await_lock);
+
+    pthread_mutex_lock(&awa.cond_lock);
+    while (!awa.stanza)
+    {
+        int code;
+
+        if (timeout <= 0)
+        {
+            pthread_cond_wait(&awa.condition, &awa.cond_lock);
+            continue;
+        }
+        code = pthread_cond_timedwait(&awa.condition, &awa.cond_lock, &ts);
+        if (code == ETIMEDOUT)
+        {
+            /* Timeout detected, give up regardless of the status of our 
+             * search. */
+            HashMapDelete(await_table, identifier);
+            pthread_mutex_unlock(&await_lock);
+            break;
+        }
+        if (code == EINVAL)
+        {
+            Log(LOG_ERR, "=========== Achievement GET! ===========");
+            Log(LOG_ERR, "%s: TIMEDWAIT RETURNED EINVAL.", __func__);
+            Log(LOG_ERR, "THIS IS, LET'S SAY, NOT GOOD, AND A SIGN OF ");
+            Log(LOG_ERR, "A PROGRAMMER ERROR. PLEASE KILLALL -9 PARSEE.");
+            Log(LOG_ERR, "");
+            Log(LOG_ERR, "YOU, HOWEVER, GET TO WIN AN AWARD FOR THIS.");
+            Log(LOG_ERR, "I AM SIMPLY JEALOUS OF YOU GETTING SUCH A ");
+            Log(LOG_ERR, "GOOD ERROR MESSAGE.");
+            Log(LOG_ERR, "=========== Achievement GET! ===========");
+        }
+    }
+
+    stanza = awa.stanza;
+    pthread_mutex_unlock(&awa.cond_lock);
+
+    pthread_cond_destroy(&awa.condition);
+    pthread_mutex_destroy(&awa.cond_lock);
+    return stanza;
+}

src/XMPPThread/Shove.c

@@ -0,0 +1,35 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Json.h>
+
+#include <StringStream.h>
+
+HashMap *
+ShoveStanza(HashMap *content, XMLElement *stanza)
+{
+    char *encoded_stanza = NULL;
+    Stream *str_writer = StrStreamWriter(&encoded_stanza);
+    if (!stanza || !content)
+    {
+        StreamClose(str_writer);
+        return content;
+    }
+
+    XMLEncode(str_writer, stanza);
+    StreamFlush(str_writer);
+    StreamClose(str_writer);
+
+    JsonValueFree(HashMapSet(content, 
+        "at.kappach.at.parsee.stanza",
+        JsonValueString(encoded_stanza)
+    ));
+    JsonValueFree(HashMapSet(content, 
+        "at.kappach.at.parsee.id",
+        JsonValueString(HashMapGet(stanza->attrs, "id"))
+    ));
+
+    Free(encoded_stanza);
+    return content;
+}
+

src/XMPPThread/Stanzas/IQ.c

@@ -0,0 +1,447 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Base64.h>
+#include <Cytoplasm/Str.h>
+#include <Cytoplasm/Log.h>
+
+#include <StringStream.h>
+#include <Matrix.h>
+#include <AS.h>
+
+#include <string.h>
+#include <ctype.h>
+
+static XMLElement *
+CreateTagWithText(const char *tn, char *text)
+{
+    XMLElement *tag = XMLCreateTag((char *) tn);
+    XMLElement *tex = XMLCreateText(text);
+
+    XMLAddChild(tag, tex);
+
+    return tag;
+}
+
+char *
+TrimBase64(char *b64)
+{
+    char *ret, *tmp;
+
+    ret = NULL;
+    while (*b64)
+    {
+        char ch[2] = { *b64, 0 };
+        if (isspace(*b64))
+        {
+            b64++;
+            continue;
+        }
+
+        tmp = ret;
+        ret = StrConcat(2, ret, ch);
+        Free(tmp);
+        b64++;
+    }
+    return ret;
+}
+
+#define DISCO   "http://jabber.org/protocol/disco#info"
+static XMLElement *
+IQGenerateQuery(void)
+{
+    XMLElement *query = XMLCreateTag("query");
+    XMLAddAttr(query, "xmlns", DISCO);
+    {
+        XMLElement *feature;
+#define IdentitySimple(c,t,n)  do \
+                        { \
+                            feature = XMLCreateTag("identity"); \
+                            XMLAddAttr(feature, "category", c); \
+                            XMLAddAttr(feature, "type", t); \
+                            XMLAddAttr(feature, "name", n); \
+                            XMLAddChild(query, feature); \
+                        } \
+                        while (0);
+        IQ_IDENTITY
+#undef IdentitySimple
+#define AdvertiseSimple(f)  do \
+                        { \
+                            feature = XMLCreateTag("feature"); \
+                            XMLAddAttr(feature, "var", f); \
+                            XMLAddChild(query, feature); \
+                        } \
+                        while (0);
+
+        IQ_ADVERT
+#undef AdvertiseSimple
+    }
+
+    return query;
+}
+void
+IQDiscoGet(ParseeData *args, XMPPComponent *jabber, XMLElement *stanza)
+{
+    char *from, *to, *id;
+    XMLElement *iq_reply, *query;
+
+    from = HashMapGet(stanza->attrs, "from");
+    to = HashMapGet(stanza->attrs, "to");
+    id = HashMapGet(stanza->attrs, "id");
+
+    /* Generate an IQ reply with discovery information */
+    iq_reply = XMLCreateTag("iq");
+    XMLAddAttr(iq_reply, "to", from);
+    XMLAddAttr(iq_reply, "from", to);
+    XMLAddAttr(iq_reply, "type", "result");
+    XMLAddAttr(iq_reply, "id", id);
+
+    query = IQGenerateQuery();
+    {
+        char *ver = XMPPGenerateVer();
+        char *node = StrConcat(3, REPOSITORY, "#", ver);
+        XMLAddAttr(query, "node", node);
+
+        Free(node);
+        Free(ver);
+    }
+    XMLAddChild(iq_reply, query);
+
+    pthread_mutex_lock(&jabber->write_lock);
+    XMLEncode(jabber->stream, iq_reply);
+    StreamFlush(jabber->stream);
+    pthread_mutex_unlock(&jabber->write_lock);
+
+    XMLFreeElement(iq_reply);
+}
+
+void
+IQResult(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    XMLElement *vcard = XMLookForTKV(stanza, "vCard", "xmlns", "vcard-temp");
+
+    XMLElement *event = XMLookForTKV(stanza, "pubsub", 
+        "xmlns", "http://jabber.org/protocol/pubsub"
+    );
+
+    (void) thr;
+
+    if (event)
+    {
+        size_t i;
+        XMLElement *retrieve = 
+            XMLookForTKV(event, "items", "node", "urn:xmpp:avatar:data");
+        if (retrieve)
+        {
+            for (i = 0; i < ArraySize(retrieve->children); i++)
+            {
+                XMLElement *item = 
+                    ArrayGet(retrieve->children, i);
+                XMLElement *avatar_data = XMLookForTKV(
+                    item, "data", "xmlns", "urn:xmpp:avatar:data"
+                );
+                XMLElement *data = ArrayGet(avatar_data->children, 0);
+                char *id = HashMapGet(item->attrs, "id");
+                char *from = HashMapGet(stanza->attrs, "from");
+                char *base64;
+                char *bdata;
+                size_t length, b64len;
+                Stream *datastream;
+                char *mxc, *from_matrix, *jid;
+                DbRef *avatars;
+                HashMap *json;
+
+                if (!data || !data->data)
+                {
+                    return;
+                }
+                avatars = DbLock(args->db, 1, "avatars");
+                if (!avatars)
+                {
+                    avatars = DbCreate(args->db, 1, "avatars");
+                }
+                json = DbJson(avatars);
+
+                if (StrEquals(GrabString(json, 1, from), id))
+                {
+                    DbUnlock(args->db, avatars);
+                    return;
+                }
+
+                base64 = TrimBase64(data->data);
+                b64len = base64 ? strlen(base64) : 0;
+                length = Base64DecodedSize(base64, b64len);
+
+                /* TODO: Bound checks for a size limit. */
+                bdata = (char *) Base64Decode(base64, b64len);
+                datastream = StrStreamReaderN(bdata, length);
+
+                /* XEP-0084 effectively assumes an image/png MIME. */
+                mxc = ASUpload(args->config, datastream, length, "image/png");
+                
+                jid = ParseeLookupJID(from);
+                from_matrix = ParseeEncodeJID(args->config, jid, false);
+                ASSetAvatar(args->config, from_matrix, mxc);
+
+                JsonValueFree(JsonSet(
+                    json, JsonValueString(id),
+                    1, from
+                ));
+                DbUnlock(args->db, avatars);
+
+                Free(mxc);
+                Free(jid);
+                Free(bdata);
+                Free(from_matrix);
+                Free(base64);
+                StreamClose(datastream);
+            }
+
+        }
+    }
+    if (vcard)
+    {
+        XMLElement *photo = XMLookForUnique(vcard, "PHOTO");
+
+        if (photo)
+        {
+            XMLElement *binval = XMLookForUnique(photo, "BINVAL");
+            XMLElement *type = XMLookForUnique(photo, "TYPE");
+            XMLElement *data = 
+                binval ? ArrayGet(binval->children, 0) : NULL;
+            XMLElement *tdata = type ? ArrayGet(type->children, 0) : NULL;
+            char *base64;
+            char *bdata;
+            size_t length, b64len;
+            Stream *datastream;
+            char *mxc = NULL, *from_matrix = NULL, *jid = NULL;
+            char *room = NULL, *resource;
+            char *from = HashMapGet(stanza->attrs, "from");
+            if (!data || !data->data)
+            {
+                Log(LOG_ERR, "%s NOT FOUND", HashMapGet(stanza->attrs, "from"));
+                return;
+            }
+
+            /* Get the base64, decode it, and shove it in a nicely put 
+             * MXC address */
+            base64 = data->data;
+            b64len = base64 ? strlen(base64) : 0;
+            length = Base64DecodedSize(base64, b64len);
+
+            bdata = Base64Decode(base64, b64len);
+            datastream = StrStreamReaderN(bdata, length);
+            mxc = ASUpload(
+                args->config, 
+                datastream, 
+                length,
+                tdata ? tdata->data : NULL
+            );
+            Free(bdata);
+            StreamClose(datastream);
+
+            room = ParseeGetBridgedRoom(args, stanza);
+            jid = ParseeLookupJID(HashMapGet(stanza->attrs, "from"));
+
+            resource = ParseeTrimJID(from);
+
+            /* TODO: More reliable system for telling the difference appart */
+            if (jid && !StrEquals(from, resource))
+            {
+                from_matrix = ParseeEncodeJID(args->config, jid, false);
+                ASSetAvatar(args->config, from_matrix, mxc);
+            }
+            else if (room)
+            {
+                char *mask = ParseeMXID(args);
+                HashMap *obj = HashMapCreate();
+
+                HashMapSet(obj, "url", JsonValueString(mxc));
+                ASSetState(
+                    args->config, 
+                    room, "m.room.avatar", "", 
+                    mask, obj
+                );
+                Free(mask);
+            }
+            
+            Free(from_matrix);
+            Free(resource);
+            Free(room);
+            Free(jid);
+            Free(mxc);
+        }
+    }
+}
+
+bool
+IQIsCommandList(ParseeData *args, XMLElement *stanza)
+{
+    char *parsee = NULL;
+    XMLElement *query = XMLookForTKV(
+        stanza, "query", "xmlns",
+        "http://jabber.org/protocol/disco#items"
+    );
+    bool ret = false;
+
+    if (!query || 
+        !StrEquals(HashMapGet(query->attrs, "node"),
+        "http://jabber.org/protocol/commands"))
+    {
+        return false;
+    }
+
+    parsee = ParseeJID(args);
+    ret = StrEquals(HashMapGet(stanza->attrs, "to"), parsee);
+    Free(parsee);
+
+    return ret;
+}
+void
+IQGet(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    XMPPComponent *jabber = args->jabber;
+    char *from = HashMapGet(stanza->attrs, "from");
+    char *to = HashMapGet(stanza->attrs, "to");
+    char *id = HashMapGet(stanza->attrs, "id");
+
+    (void) thr;
+
+    if (IQIsCommandList(args, stanza))
+    {
+        XMLElement *iq_reply = XMLCreateTag("iq");
+        XMLAddAttr(iq_reply, "type", "result");
+        XMLAddAttr(iq_reply, "from", to);
+        XMLAddAttr(iq_reply, "to", from);
+        XMLAddAttr(iq_reply, "id", id);
+        {
+            XMLElement *q = XMLCreateTag("query");
+            XMLAddAttr(q, "xmlns", "http://jabber.org/protocol/disco#items");
+            XMLAddAttr(q, "node", "http://jabber.org/protocol/commands");
+            XMPPShoveCommandList(thr->info->m, to, q);
+            XMLAddChild(iq_reply, q);
+        }
+        pthread_mutex_lock(&jabber->write_lock);
+        XMLEncode(jabber->stream, iq_reply);
+        StreamFlush(jabber->stream);
+        pthread_mutex_unlock(&jabber->write_lock);
+        XMLFreeElement(iq_reply);
+    }
+    else if (XMLookForTKV(stanza, "vCard", "xmlns", "vcard-temp"))
+    {
+        Log(LOG_INFO, "vCard information GET for %s", to);
+
+        /* TODO: "a compliant server MUST respond on behalf of the 
+         * requestor and not forward the IQ to the requestee's 
+         * connected resource". */
+        if (!strncmp(to, "parsee@", 7))
+        {
+            XMLElement *iqVCard = XMLCreateTag("iq");
+            XMLAddAttr(iqVCard, "from", to);
+            XMLAddAttr(iqVCard, "to", from);
+            XMLAddAttr(iqVCard, "id", id);
+            XMLAddAttr(iqVCard, "type", "result");
+            {
+                XMLElement *vCard = XMLCreateTag("vCard");
+                XMLAddAttr(vCard, "xmlns", "vcard-temp");
+                {
+                    XMLElement *fn = CreateTagWithText(
+                        "FN", "Parsee Mizuhashi"
+                    );
+                    XMLElement *nick = CreateTagWithText(
+                        "NICKNAME", "Parsee"
+                    );
+                    XMLElement *url = CreateTagWithText(
+                        "URL", REPOSITORY
+                    );
+                    /* TODO: Maybe abstract the vCard code. */
+                    XMLAddChild(vCard, nick);
+                    XMLAddChild(vCard, url);
+                    XMLAddChild(vCard, fn);
+                }
+                XMLAddChild(iqVCard, vCard);
+            }
+
+            pthread_mutex_lock(&jabber->write_lock);
+            XMLEncode(jabber->stream, iqVCard);
+            StreamFlush(jabber->stream);
+            pthread_mutex_unlock(&jabber->write_lock);
+            XMLFreeElement(iqVCard);
+        }
+    }
+    else if (XMLookForTKV(stanza, "query", "xmlns", DISCO))
+    {
+        IQDiscoGet(args, jabber, stanza);
+    }
+    else if (XMLookForTKV(stanza, "query", "xmlns", "jabber:iq:version"))
+    {
+        XMLElement *iq_reply, *query;
+        XMLElement *name, *version;
+
+        iq_reply = XMLCreateTag("iq");
+        XMLAddAttr(iq_reply, "to", from);
+        XMLAddAttr(iq_reply, "from", to);
+        XMLAddAttr(iq_reply, "type", "result");
+        XMLAddAttr(iq_reply, "id", id);
+
+        query = XMLCreateTag("query");
+        XMLAddAttr(query, "xmlns", "jabber:iq:version");
+        {
+            name = XMLCreateTag("name");
+            version = XMLCreateTag("version");
+
+            XMLAddChild(name, XMLCreateText(NAME));
+            XMLAddChild(version, XMLCreateText(VERSION));
+        }
+        XMLAddChild(query, name);
+        XMLAddChild(query, version);
+        XMLAddChild(iq_reply, query);
+
+        pthread_mutex_lock(&jabber->write_lock);
+        XMLEncode(jabber->stream, iq_reply);
+        StreamFlush(jabber->stream);
+        pthread_mutex_unlock(&jabber->write_lock);
+        XMLFreeElement(iq_reply);
+    }
+    else
+    {
+        Log(LOG_WARNING, "Unknown I/Q received:");
+        XMLEncode(StreamStdout(), stanza);
+        StreamPrintf(StreamStdout(),"\n");
+        StreamFlush(StreamStdout());
+    }
+
+}
+void
+IQError(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    /* TODO */
+}
+void
+IQSet(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    XMPPCommandManager *manager = thr->info->m;
+    XMPPManageCommand(manager, stanza, args);
+}
+#undef DISCO
+
+void IQStanza(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    char *type;
+    type = HashMapGet(stanza->attrs, "type");
+#define OnType(ctyp, callback)  do \
+                                { \
+                                    if (StrEquals(type, #ctyp)) \
+                                    { \
+                                        callback(args, stanza, thr); \
+                                        return; \
+                                    } \
+                                } \
+                                while (0)
+    
+    OnType(get, IQGet);
+    OnType(set, IQSet);
+    OnType(error, IQError);
+    OnType(result, IQResult);
+#undef OnType
+}

src/XMPPThread/Stanzas/Message.c

@@ -0,0 +1,375 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Str.h>
+
+#include <Matrix.h>
+#include <AS.h>
+
+#include <string.h>
+
+bool
+MessageStanza(ParseeData *args, XMLElement *stanza, XMPPThread *thr)
+{
+    XMPPComponent *jabber = args->jabber;
+
+    XMLElement *reactions = NULL;
+    XMLElement *body = NULL;
+    XMLElement *data = NULL;
+    XMLElement *event = NULL;
+
+    char *to, *room, *from, *from_matrix, *decode_from;
+    char *mroom_id = NULL;
+    char *replaced  = XMPPGetReplacedID(stanza);
+    char *retracted = XMPPGetRetractedID(stanza);
+    char *reply_to  = XMPPGetReply(stanza);
+    char *moderated = XMPPGetModeration(stanza);
+    size_t i;
+
+    to = NULL;
+    from = NULL;
+    decode_from = NULL;
+    from_matrix = NULL;
+
+    from = HashMapGet(stanza->attrs, "from");
+    if (ParseeManageBan(args, from, NULL))
+    {
+        XMLFreeElement(stanza);
+        return false;
+    }
+
+    if (moderated)
+    {
+        /* TODO: Parsee MUST check if it is a valid MUC */
+        char *resource = ParseeGetResource(from);
+        char *chat_id = ParseeGetFromMUCID(args, from);
+        char *event_id  = NULL;
+        char *encoded = ParseeMXID(args);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
+        if (!resource && chat_id)
+        {
+            event_id = ParseeGetEventFromID(args, stanza, moderated);
+            ASRedact(args->config, mroom_id, NULL, event_id);
+            ParseePushAllStanza(args, stanza, event_id);
+            pthread_mutex_unlock(&thr->info->chk_lock);
+
+            Free(event_id);
+        }
+        Free(mroom_id);
+        Free(resource);
+        Free(encoded);
+        Free(chat_id);
+    }
+    body = XMLookForUnique(stanza, "body");
+
+    event = XMLookForTKV(stanza, "event", 
+        "xmlns", "http://jabber.org/protocol/pubsub#event"
+    );
+    if (event)
+    {
+        size_t i;
+        XMLElement *items = 
+            XMLookForTKV(event, "items", "node", "urn:xmpp:avatar:metadata");
+        if (items)
+        {
+            for (i = 0; i < ArraySize(items->children); i++)
+            {
+                ManageProfileItem(
+                    args, ArrayGet(items->children, i), 
+                    stanza, thr
+                );
+            }
+        }
+    }
+
+#define CHAT_STATES "http://jabber.org/protocol/chatstates"
+    if (XMLookForTKV(stanza, "composing", "xmlns", CHAT_STATES))
+    {
+        decode_from = ParseeLookupJID(from);
+        from_matrix = ParseeEncodeJID(args->config, decode_from, true);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
+        
+        ASType(args->config, from_matrix, mroom_id, true);
+
+        Free(decode_from);
+        Free(from_matrix);
+        Free(mroom_id);
+        mroom_id = NULL;
+        decode_from = NULL;
+        from_matrix = NULL;
+    }
+    if (XMLookForTKV(stanza, "active", "xmlns", CHAT_STATES))
+    {
+        char *latest = NULL;
+        decode_from = ParseeLookupJID(from);
+        from_matrix = ParseeEncodeJID(args->config, decode_from, true);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
+
+        latest = ParseeLookupHead(mroom_id);
+
+        ASType(args->config, from_matrix, mroom_id, false);
+        ASPresence(args->config, from_matrix, mroom_id, latest);
+
+        Free(decode_from);
+        Free(from_matrix);
+        Free(latest);
+        Free(mroom_id);
+        mroom_id = NULL;
+        decode_from = NULL;
+        from_matrix = NULL;
+    }
+    if (XMLookForTKV(stanza, "paused", "xmlns", CHAT_STATES) ||
+        XMLookForTKV(stanza, "received", "xmlns", "urn:xmpp:receipts") ||
+        XMLookForTKV(stanza, "displayed", "xmlns", "urn:xmpp:chat-markers:0"))
+    {
+        /* TODO: Use stanza ID if possible */
+        char *latest = NULL;
+        decode_from = ParseeLookupJID(from);
+        from_matrix = ParseeEncodeJID(args->config, decode_from, true);
+        mroom_id = ParseeGetBridgedRoom(args, stanza);
+
+        latest = ParseeLookupHead(mroom_id);
+
+        ASPresence(args->config, from_matrix, mroom_id, latest);
+
+        Free(decode_from);
+        Free(from_matrix);
+        Free(latest);
+        Free(mroom_id);
+        mroom_id = NULL;
+        decode_from = NULL;
+        from_matrix = NULL;
+
+    }
+#undef CHAT_STATES
+    if (!body)
+    {
+        XMLFreeElement(stanza);
+        return false;
+    }
+
+    /* TODO: On semi-anonymous MUCs, it might be preferable to use a 
+     * form of the occupant ID as the base, as it is more unique, and 
+     * less prone to trigger the character limit on Matrix. 
+     *
+     * See: https://xmpp.org/extensions/xep-0421.html */
+    to = ParseeDecodeMXID(HashMapGet(stanza->attrs, "to"));
+    decode_from = ParseeLookupJID(from);
+    from_matrix = ParseeEncodeJID(args->config, decode_from, true);
+    room = ParseeFindDMRoom(args, to, from);
+    data = ArrayGet(body->children, 0);
+
+    /* TODO: CLEAN THAT UP */
+    mroom_id = ParseeGetBridgedRoom(args, stanza);
+    if (!mroom_id && !room && !XMPPIsParseeStanza(stanza) &&
+        to && *to == '@')
+    {
+        DbRef *room_ref;
+        HashMap *room_json;
+        char *from = HashMapGet(stanza->attrs, "from");
+
+        ASRegisterUser(args->config, from_matrix);
+        room = ASCreateDM(args->config, from_matrix, to);
+        mroom_id = StrDuplicate(room);
+        if (room)
+        {
+            room_ref = DbCreate(args->db, 3, "rooms", room, "data");
+            room_json = DbJson(room_ref);
+            HashMapSet(room_json, "is_direct", JsonValueBoolean(true));
+            HashMapSet(room_json, "xmpp_user", JsonValueString(from));
+            DbUnlock(args->db, room_ref);
+            ParseePushDMRoom(args, to, from, room);
+        }
+    }
+
+    /* TODO: THIS IS A HACK. THIS CODE IGNORES ALL MUC MESSAGES EVER 
+     * SENT TO NON-PARSEE PUPPETS, AS A "FIX" TO THE MULTITHREADED 
+     * ISSUE. 
+     *
+     * I HATE THIS. I NEED TO FIND A BETTER WAY. */
+    if (!room)
+    {
+        if (strncmp(HashMapGet(stanza->attrs, "to"), "parsee@", 7))
+        {
+            goto end;
+        }
+    }
+
+    if (mroom_id && !XMPPIsParseeStanza(stanza))
+    {
+        char *res = ParseeGetResource(from);
+        char *encoded   = ParseeEncodeJID(args->config, decode_from, false);
+        char *event_id  = NULL;
+        bool chat       = false;
+
+        if (StrEquals(HashMapGet(stanza->attrs, "type"), "chat"))
+        {
+            Free(encoded);
+            encoded = StrDuplicate(from_matrix);
+            chat = true;
+        }
+
+        {
+            char *parsee = ParseeJID(args);
+
+            /* Subscribe to the sender's metadata node. */
+            XMLElement *ps = CreatePubsubRequest(
+                parsee, decode_from, "urn:xmpp:avatar:metadata"
+            );
+            pthread_mutex_lock(&jabber->write_lock);
+            XMLEncode(jabber->stream, ps);
+            StreamFlush(jabber->stream);
+            pthread_mutex_unlock(&jabber->write_lock);
+
+            XMLFreeElement(ps);
+            Free(parsee);
+        }
+
+        pthread_mutex_lock(&thr->info->chk_lock);
+        if (ParseeVerifyAllStanza(args, stanza) && !replaced)
+        {
+            XMLElement *oob, *oob_data;
+
+            pthread_mutex_unlock(&thr->info->chk_lock);
+            ASRegisterUser(args->config, encoded);
+            if (!chat)
+            {
+                ASSetName(args->config, encoded, res);
+            }
+            ASInvite(args->config, mroom_id, encoded);
+            Free(ASJoin(args->config, mroom_id, encoded));
+
+            /* Check if it is a media link */
+            oob = XMLookForTKV(stanza, "x", "xmlns", "jabber:x:oob");
+            reactions = XMLookForTKV(stanza,
+                "reactions", "xmlns", "urn:xmpp:reactions:0"
+            );
+            if (oob)
+            {
+                char *mxc, *mime = NULL;
+                HashMap *content = NULL;
+                oob_data = XMLookForUnique(oob, "url");
+                oob_data = 
+                    oob_data ? ArrayGet(oob_data->children, 0) : NULL;
+
+                if (oob_data)
+                {
+                    mxc = ASReupload(args->config, oob_data->data, &mime);
+                    content = MatrixCreateMedia(mxc, data->data, mime);
+
+                    /* Yeah, no, I'm not modifying the media creation code. */
+                    HashMapSet(content, 
+                        "at.kappach.at.parsee.external", 
+                        JsonValueString(oob_data->data)
+                    );
+                    ShoveStanza(content, stanza);
+
+                    event_id = ASSend(
+                        args->config, mroom_id, encoded, 
+                        "m.room.message", content
+                    );
+                    Free(mime);
+                    Free(mxc);
+                }
+            }
+            else if (reactions)
+            {
+                Array *react_child = reactions->children;
+                size_t reacts = ArraySize(react_child);
+                event_id = ParseeGetReactedEvent(args, stanza);
+                for (i = 0; i < reacts; i++)
+                {
+                    XMLElement *reaction, *react_data;
+                    reaction = ArrayGet(react_child, i);
+                    react_data = ArrayGet(reaction->children, 0);
+
+                    /* TODO: We should manage removed reactions. */
+                    Free(ASSend(
+                        args->config, mroom_id, encoded, 
+                        "m.reaction", 
+                        ShoveStanza(
+                            MatrixCreateReact(event_id, react_data->data),
+                            stanza
+                        )
+                    ));
+                }
+                Free(event_id);
+                event_id = NULL;
+            }
+            else if (retracted)
+            {
+                event_id = ParseeGetEventFromID(args, stanza, retracted);
+                ASRedact(args->config, mroom_id, encoded, event_id);
+                ParseePushAllStanza(args, stanza, event_id);
+                pthread_mutex_unlock(&thr->info->chk_lock);
+
+                Free(event_id);
+                event_id = NULL;
+            }
+            else
+            {
+                /* TODO: Use HTML-formatted bodies, and respect the fallback 
+                 * trims the stanza provides us if possible. Element does not 
+                 * like raw bodies on replies too. Go figure. */
+                size_t off = 
+                    reply_to ? ParseeFindDatastart(data->data) : 0;
+                HashMap *ev = MatrixCreateMessage(data->data + off);
+                if (reply_to)
+                {
+                    char *reply_id = ParseeGetEventFromID(args, stanza, reply_to);
+                    MatrixSetReply(ev, reply_id);
+                    Free(reply_id);
+                }
+                ShoveStanza(ev, stanza);
+                event_id = ASSend(
+                    args->config, mroom_id, encoded, 
+                    "m.room.message", ev
+                );
+            }
+            pthread_mutex_lock(&thr->info->chk_lock);
+            ParseePushAllStanza(args, stanza, event_id);
+            Free(event_id);
+            pthread_mutex_unlock(&thr->info->chk_lock);
+        }
+        else if (replaced)
+        {
+            event_id = ParseeGetEventFromID(args, stanza, replaced);
+            Free(ASSend(
+                args->config, mroom_id, encoded, 
+                "m.room.message", 
+                ShoveStanza(MatrixCreateReplace(event_id, data->data), stanza)
+            ));
+            ParseePushAllStanza(args, stanza, event_id);
+            pthread_mutex_unlock(&thr->info->chk_lock);
+
+            Free(event_id);
+        }
+        else
+        {
+            pthread_mutex_unlock(&thr->info->chk_lock);
+        }
+        Free(res);
+        Free(encoded);
+    }
+    else
+    {
+        XMLElement *parsee = XMLookForUnique(stanza, "x-parsee");
+        XMLElement *event = XMLookForUnique(parsee, "event-id");
+        XMLElement *e_d = ArrayGet(event ? event->children : NULL, 0);
+
+        if (e_d)
+        {
+            ParseePushAllStanza(args, stanza, e_d->data);
+        }
+    }
+end:
+    Free(mroom_id);
+    mroom_id = NULL;
+    Free(from_matrix);
+    Free(decode_from);
+    Free(room);
+    Free(to);
+
+    return true;
+}
+

src/XMPPThread/Stanzas/Presence.c

@@ -0,0 +1,258 @@
+#include "XMPPThread/internal.h"
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Str.h>
+#include <Cytoplasm/Log.h>
+
+#include <Matrix.h>
+#include <AS.h>
+
+#include <string.h>
+
+static XMLElement *
+CreateVCardRequest(char *from, char *to)
+{
+    XMLElement *vcard_request, *vcard;
+    char *id;
+    vcard_request = XMLCreateTag("iq");
+    XMLAddAttr(vcard_request, "from", from);
+    XMLAddAttr(vcard_request, "to", to);
+    XMLAddAttr(vcard_request, "id", (id = StrRandom(16)));
+    XMLAddAttr(vcard_request, "type", "get");
+
+    vcard = XMLCreateTag("vCard");
+    XMLAddAttr(vcard, "xmlns", "vcard-temp");
+    XMLAddChild(vcard_request, vcard);
+
+    Free(id);
+
+    return vcard_request;
+}
+
+static UserStatus
+GuessStatus(XMLElement *stanza)
+{
+    /* C.F RFC3921: XMPP IM */
+    XMLElement *show = XMLookForUnique(stanza, "show");
+    XMLElement *data = show ? ArrayGet(show->children, 0) : NULL;
+
+    if (!show || !data)
+    {
+        return USER_STATUS_ONLINE;
+    }
+
+    if (StrEquals(data->data, "away") ||
+        StrEquals(data->data, "xa"))
+    {
+        return USER_STATUS_OFFLINE;
+    }
+    if (StrEquals(data->data, "chat"))
+    {
+        return USER_STATUS_ONLINE;
+    }
+    if (StrEquals(data->data, "dnd"))
+    {
+        return USER_STATUS_UNAVAILABLE;
+    }
+    return USER_STATUS_ONLINE;
+}
+void
+PresenceStanza(ParseeData *args, XMLElement *stanza)
+{
+#define MUC_USER_NS "http://jabber.org/protocol/muc#user"
+    XMLElement *user_info;
+    XMLElement *vc = XMLookForTKV(stanza, "x", "xmlns", "vcard-temp:x:update");
+    XMLElement *status = XMLookForUnique(stanza, "status");
+    char *oid = HashMapGet(stanza->attrs, "from");
+
+    if ((user_info = XMLookForTKV(stanza, "x", "xmlns", MUC_USER_NS)))
+    {
+        XMLElement *item = XMLookForUnique(user_info, "item");
+        XMLElement *status = XMLookForUnique(user_info, "status");
+#define IsStatus(code) (status && \
+                        StrEquals(HashMapGet(status->attrs, "code"), #code))
+        char *jid = item ? HashMapGet(item->attrs, "jid") : NULL;
+        char *trim = ParseeTrimJID(jid);
+        char *from = NULL;
+        char *type = HashMapGet(stanza->attrs, "type");
+        char *room = ParseeGetBridgedRoom(args, stanza);
+        char *decode_from = ParseeLookupJID(oid);
+        char *real_matrix = ParseeDecodeMXID(decode_from);
+        char *matrix_user_pl = ParseeEncodeJID(args->config, trim, false);
+        char *affiliation = HashMapGet(item->attrs, "affiliation");
+        int power_level = 0;
+        char *parsee = ParseeMXID(args);
+
+        Free(trim);
+
+        if (!real_matrix || *real_matrix != '@')
+        {
+            Free(real_matrix);
+            real_matrix = ParseeEncodeJID(args->config, decode_from, false);
+        }
+
+        if (StrEquals(affiliation, "owner"))
+        {
+            power_level = 98;
+        }
+        else if (StrEquals(affiliation, "admin"))
+        {
+            power_level = 50;
+        }
+        else if (StrEquals(affiliation, "member"))
+        {
+            /* TODO: Reconsider setting the PL if a member, as there's no 
+             * clear reason to do this in some cases. */
+            power_level = 0;
+        }
+        else if (StrEquals(affiliation, "outcast"))
+        {
+            power_level = -1;
+        }
+
+        /* Set the user's PL
+         * TODO: Do NOT change the PL of *real* people nilly-willy.
+         * In some scenarios, this is really bad behaviour. */
+        if (room)
+        {
+            HashMap *powers = ASGetPL(args->config, room);
+            HashMap *users = GrabObject(powers, 1, "users");
+            int64_t level = GrabInteger(powers, 2, "users", matrix_user_pl);
+
+            /* I may or may not have fucked up the state hard with this
+             * lacking any checking. (--gen) */
+            if (StrEquals(parsee, matrix_user_pl))
+            {
+                /* TODO: Give the user an achievement, this is goofy. */
+                Log(LOG_ERR, "BAD PL DOWNGRADE (%d->%d)", level, power_level);
+            }
+            if (users && level != power_level && 
+                !StrEquals(parsee, matrix_user_pl) &&
+                matrix_user_pl)
+            {
+                JsonValue *val = JsonValueInteger(power_level);
+                JsonValueFree(JsonSet(users, val, 1, matrix_user_pl));
+                ASSetPL(args->config, room, powers);
+            }
+            else
+            {
+                JsonFree(powers);
+            }
+        }
+
+
+        if (StrEquals(type, "unavailable") && !StrEquals(parsee, real_matrix))
+        {
+            /* If not an MXID, use the Parsee user */
+            if (IsStatus(301))
+            {
+                ASBan(args->config, room, real_matrix);
+            }
+            else if (IsStatus(307))
+            {
+                ASKick(args->config, room, real_matrix);
+            }
+            else if (IsStatus(303))
+            {
+                char *nick = HashMapGet(item->attrs, "nick");
+                ASSetName(args->config, real_matrix, nick);
+            }
+            else
+            {
+                ASLeave(args->config, room, real_matrix);
+            }
+        }
+
+        if (jid)
+        {
+            ParseePushJIDTable(oid, jid);
+        }
+
+        Free(from);
+        Free(decode_from);
+        Free(real_matrix);
+        Free(matrix_user_pl);
+        Free(parsee);
+        Free(room);
+    }
+    if (status)
+    {
+        XMLElement *status_data = ArrayGet(status->children, 0);
+        char *decode_from = ParseeLookupJID(oid);
+        char *trimmed = ParseeTrimJID(decode_from);
+        char *from_matrix = ParseeEncodeJID(args->config, trimmed, false);
+        char *status_str = NULL;
+        if (status_data)
+        {
+            status_str = status_data->data;
+        }
+
+        /* TODO: "The server will automatically set a user's presence to 
+         * unavailable if their last active time was over a threshold value 
+         * (e.g. 5 minutes)."
+         * We _will_ need to manage those cases properly(cronjob?) if we want 
+         * XMPP presences to sync properly */
+        ASSetStatus(
+            args->config, from_matrix, 
+            GuessStatus(stanza), status_str
+        );
+
+        Free(decode_from);
+        Free(from_matrix);
+        Free(trimmed);
+    }
+    if (vc)
+    {
+        XMLElement *photo = XMLookForUnique(vc, "photo");
+        XMLElement *p_dat = photo ? ArrayGet(photo->children, 0) : NULL;
+        XMLElement *vcard_request;
+        XMPPComponent *jabber = args->jabber;
+        char *from;
+
+        DbRef *avatars;
+        HashMap *json;
+        char *avatar_id;
+
+        if (!p_dat)
+        {
+            return;
+        }
+
+        avatars = DbLock(args->db, 1, "avatars");
+        if (!avatars)
+        {
+            avatars = DbCreate(args->db, 1, "avatars");
+        }
+        json = DbJson(avatars);
+
+        avatar_id = GrabString(json, 1, oid);
+        if (avatar_id && StrEquals(avatar_id, p_dat->data))
+        {
+            DbUnlock(args->db, avatars);
+            return;
+        }
+        JsonValueFree(JsonSet(
+            json, JsonValueString(p_dat->data),
+            1, oid)
+        );
+
+        DbUnlock(args->db, avatars);
+
+        from = ParseeJID(args);
+
+        vcard_request = CreateVCardRequest(
+            from, HashMapGet(stanza->attrs, "from")
+        );
+        pthread_mutex_lock(&jabber->write_lock);
+        XMLEncode(jabber->stream, vcard_request);
+        StreamFlush(jabber->stream);
+        pthread_mutex_unlock(&jabber->write_lock);
+
+        XMLFreeElement(vcard_request);
+        Free(from);
+    }
+
+    /* TODO: Sending VCard on presence is slow and stalls the thread */
+#undef MUC_USER_NS
+}
+

src/XMPPThread/XMPPThread.c

@@ -0,0 +1,381 @@
+#include <Parsee.h>
+
+#include <pthread.h>
+#include <string.h>
+#include <ctype.h>
+
+#include <Cytoplasm/Memory.h>
+#include <Cytoplasm/Base64.h>
+#include <Cytoplasm/Util.h>
+#include <Cytoplasm/Log.h>
+#include <Cytoplasm/Str.h>
+#include <Cytoplasm/Sha.h>
+
+#include <StringStream.h>
+#include <XMPPCommand.h>
+#include <Matrix.h>
+#include <XMPP.h>
+#include <XML.h>
+#include <AS.h>
+
+#include "XMPPThread/internal.h"
+
+/* Generates a SHA-256 hash of the ver field. */
+char *
+XMPPGenerateVer(void)
+{
+    char *S = NULL;
+    unsigned char *Sha = NULL;
+    Array *identities = ArrayCreate();
+    Array *features = ArrayCreate();
+    size_t i;
+
+    /* Initialise identity table, to be sorted */
+#define IdentitySimple(cat, Type, Name) { \
+    XMPPIdentity *id = Malloc(sizeof(*id)); \
+    id->category = cat; \
+    id->lang = NULL; \
+    id->type = Type; \
+    id->name = Name; \
+    ArrayAdd(identities, id); }
+    IQ_IDENTITY
+#undef IdentitySimple
+#define AdvertiseSimple(feature) ArrayAdd(features, feature);
+    IQ_ADVERT
+#undef AdvertiseSimple
+    ArraySort(identities, IdentitySort);
+    for (i = 0; i < ArraySize(identities); i++)
+    {
+        XMPPIdentity *identity = ArrayGet(identities, i);
+        char *id_chunk = StrConcat(7, 
+            identity->category, "/",
+            identity->type, "/",
+            identity->lang, "/",
+            identity->name);
+        char *tmp = S;
+        S = StrConcat(3, S, id_chunk, "<");
+        Free(tmp);
+        Free(id_chunk);
+    }
+
+    ArraySort(features, ((int (*) (void *, void *)) ICollate));
+    for (i = 0; i < ArraySize(features); i++)
+    {
+        char *feature = ArrayGet(features, i);
+        char *tmp = S;
+        S = StrConcat(3, S, feature, "<");
+        Free(tmp);
+    }
+
+    Sha = Sha1(S);
+    Free(S);
+    S = Base64Encode((const char *) Sha, 20);
+    Free(Sha);
+
+    ArrayFree(features);
+    for (i = 0; i < ArraySize(identities); i++)
+    {
+        XMPPIdentity *identity = ArrayGet(identities, i);
+        /* We don't have to do anything here. */
+        Free(identity);
+    }
+    ArrayFree(identities);
+
+    return S;
+}
+
+XMLElement *
+RetrieveStanza(XMPPThread *thread)
+{
+    XMLElement *ret = NULL;
+
+    pthread_mutex_lock(&thread->info->lock);
+    ret = ArrayDelete(thread->info->stanzas, 0);
+    pthread_mutex_unlock(&thread->info->lock);
+
+    return ret;
+}
+static void
+PushStanza(XMPPThreadInfo *info, XMLElement *stanza)
+{
+    pthread_mutex_lock(&info->lock);
+    ArrayAdd(info->stanzas, stanza);
+    pthread_mutex_unlock(&info->lock);
+}
+
+static void *
+XMPPDispatcher(void *argp)
+{
+    XMPPThread *thread = argp;
+    ParseeData *args = thread->info->args;
+
+    while (thread->info->running)
+    {
+        XMLElement *stanza = RetrieveStanza(thread);
+ 
+        /* TODO: I've seen some spikes in some threads. */
+        if (!stanza)
+        {
+            UtilSleepMillis(5);
+            continue;
+        }
+
+        if (StrEquals(stanza->name, "presence"))
+        {
+            PresenceStanza(args, stanza);
+            XMLFreeElement(stanza);
+            continue;
+        }
+        else if (StrEquals(stanza->name, "message"))
+        {
+            if (!MessageStanza(args, stanza, thread))
+            {
+                continue;
+            }
+        }
+        else if (StrEquals(stanza->name, "iq"))
+        {
+            IQStanza(args, stanza, thread);
+        }
+        else
+        {
+            Log(LOG_WARNING, "Unknown stanza '%s':", stanza->name);
+            XMLEncode(StreamStdout(), stanza);
+            StreamPrintf(StreamStdout(), "\n");
+            StreamFlush(StreamStdout());
+        }
+        XMLFreeElement(stanza);
+    }
+
+    return NULL;
+}
+
+typedef struct XMPPAwait {
+    pthread_mutex_t cond_lock;
+    pthread_cond_t condition;
+    bool usable;
+
+    XMLElement *stanza;
+} XMPPAwait;
+
+static pthread_mutex_t await_lock = PTHREAD_MUTEX_INITIALIZER;
+static HashMap *await_table = NULL;
+static XMPPThreadInfo info;
+
+size_t
+ParseeCongestion(void)
+{
+    size_t congestion;
+    pthread_mutex_lock(&info.lock);
+    congestion = ArraySize(info.stanzas);
+    pthread_mutex_unlock(&info.lock);
+
+    return congestion;
+}
+
+void *
+ParseeXMPPThread(void *argp)
+{
+    ParseeData *args = argp;
+    XMPPComponent *jabber = args->jabber;
+    XMLElement *stanza = NULL;
+    size_t i;
+        
+    /* Initialise the await table */
+    await_table = HashMapCreate();
+
+    /* Initialise the command manager, and add all ad-hoc commands */
+    info.m = XMPPCreateManager(args);
+    {
+        XMPPCommand *cmd;
+#define XMPP_COMMAND(f,n,t,s) \
+        cmd =  XMPPBasicCmd( \
+            n, t, f \
+        ); \
+        s \
+        XMPPRegisterCommand(info.m, cmd);
+        XMPPCOMMANDS
+#undef XMPP_COMMAND
+    }
+
+    /* Initialise the FIFO */
+    info.stanzas = ArrayCreate();
+    pthread_mutex_init(&info.lock, NULL);
+
+    /* ... and its readers. */
+    /* TODO: Make that configurable. */
+    info.available_dispatchers = 16;
+    info.dispatchers = Malloc(
+        sizeof(*info.dispatchers) * info.available_dispatchers
+    );
+
+    info.args = args;
+    info.jabber = jabber;
+    info.running = true;
+    pthread_mutex_init(&info.chk_lock, NULL);
+
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t *thr = &info.dispatchers[i].thr;
+        info.dispatchers[i].info = &info;
+
+        pthread_create(thr, NULL, XMPPDispatcher, &info.dispatchers[i]);
+    }
+
+    while (true)
+    {
+        char *id;
+
+        stanza = XMLDecode(jabber->stream, false);
+        if (!stanza)
+        {
+            break;
+        }
+
+        id = HashMapGet(stanza->attrs, "id");
+        if (id)
+        {
+            XMPPAwait *await;
+            /* Lock out the table to see if we're awaiting. */
+            pthread_mutex_lock(&await_lock);
+            if ((await = HashMapGet(await_table, id)))
+            {
+                pthread_mutex_lock(&await->cond_lock);
+                await->stanza = stanza;             
+                pthread_cond_broadcast(&await->condition);
+                pthread_mutex_unlock(&await->cond_lock);
+
+                HashMapDelete(await_table, id);
+
+                pthread_mutex_unlock(&await_lock);
+                continue;
+            }
+            pthread_mutex_unlock(&await_lock);
+        }
+
+        /* Push it into the stanza FIFO. A dispatcher thread should then 
+         * be able to freely grab a value(locked by a mutex). We can't make 
+         * dispatchers read stanzas on their own, since that's _not_ how 
+         * streams work, but this should mitigate some issues, and allow a 
+         * few threads to be busy, while the rest of Parsee works. */
+        PushStanza(&info, stanza);
+    }
+
+    info.running = false;
+    Log(LOG_INFO, "Joining subthreads...");
+    for (i = 0; i < info.available_dispatchers; i++)
+    {
+        pthread_t thr = info.dispatchers[i].thr;
+        pthread_join(thr, NULL);
+    }
+    Log(LOG_INFO, "Joined subthreads...");
+    Free(info.dispatchers);
+
+    for (i = 0; i < ArraySize(info.stanzas); i++)
+    {
+        XMLFreeElement(ArrayGet(info.stanzas, i));
+    }
+    ArrayFree(info.stanzas);
+
+    HashMapFree(await_table);
+
+    pthread_mutex_destroy(&info.lock);
+
+    XMPPFreeManager(info.m);
+    return NULL;
+}
+
+#include <time.h>
+#include <errno.h>
+
+/* TODO: This function does NIT behave well. */
+XMLElement *
+ParseeAwaitStanza(char *identifier, int64_t timeout)
+{
+    /* TODO: Pthreads HATE me using Malloc here, so I'm abusing stackspace.
+     * Not *too much* of a problem, just a weird oddity. If anyone has a clue 
+     * on why that happens (at least on ARM64 with a Pi4 running Debian), let 
+     * me know! */
+    XMPPAwait awa;
+    XMLElement *stanza;
+    struct timespec ts;
+    if (!identifier)
+    {
+        return NULL;
+    }
+
+    /* Convert into an absolute timeout.
+     * =================================
+     * XXX: For anyone using timespecs: MAKE ABSOLUTELY FUCKING SURE YOUR NANOS 
+     * ARE IN RANGE. THIS UNIRONICALLY CAUSED SOME REAL CONCURRENCY ERRORS ON MY 
+     * SIDE FROM TV_NSEC OVERFLOWING AND TIMEDWAIT RETURNING EINVAL. */
+    if (timeout > 0)
+    {
+        int64_t seconds = timeout / (1 SECONDS);
+        int64_t msecond = timeout % (1 SECONDS);
+
+        clock_gettime(CLOCK_REALTIME, &ts);
+
+        ts.tv_sec += seconds            ;
+        ts.tv_nsec+= msecond * 1000000  ;
+        if (ts.tv_nsec > 999999999)
+        {
+            int64_t sec_delta = ts.tv_nsec / 1000000000;
+            int64_t nsc_delta = ts.tv_nsec % 1000000000;
+
+            ts.tv_nsec = nsc_delta;
+            ts.tv_sec += sec_delta;
+        }
+
+        (void) msecond;
+    }
+
+    pthread_mutex_lock(&await_lock);
+
+    pthread_cond_init(&awa.condition, NULL);
+    pthread_mutex_init(&awa.cond_lock, NULL);
+    awa.stanza = NULL;
+    
+    HashMapSet(await_table, identifier, &awa);
+    pthread_mutex_unlock(&await_lock);
+
+    pthread_mutex_lock(&awa.cond_lock);
+    while (!awa.stanza)
+    {
+        int code;
+
+        if (timeout <= 0)
+        {
+            pthread_cond_wait(&awa.condition, &awa.cond_lock);
+            continue;
+        }
+        code = pthread_cond_timedwait(&awa.condition, &awa.cond_lock, &ts);
+        if (code == ETIMEDOUT)
+        {
+            /* Timeout detected, give up regardless of the status of our 
+             * search. */
+            HashMapDelete(await_table, identifier);
+            pthread_mutex_unlock(&await_lock);
+            break;
+        }
+        if (code == EINVAL)
+        {
+            Log(LOG_ERR, "=========== Achievement GET! ===========");
+            Log(LOG_ERR, "%s: TIMEDWAIT RETURNED EINVAL.", __func__);
+            Log(LOG_ERR, "THIS IS, LET'S SAY, NOT GOOD, AND A SIGN OF ");
+            Log(LOG_ERR, "A PROGRAMMER ERROR. PLEASE PKILL -9 PARSEE.");
+            Log(LOG_ERR, "");
+            Log(LOG_ERR, "YOU, HOWEVER, GET TO WIN AN AWARD FOR THIS.");
+            Log(LOG_ERR, "I AM SIMPLY JEALOUS OF YOU GETTING SUCH A ");
+            Log(LOG_ERR, "GOOD ERROR MESSAGE.");
+            Log(LOG_ERR, "=========== Achievement GET! ===========");
+        }
+    }
+
+    stanza = awa.stanza;
+    pthread_mutex_unlock(&awa.cond_lock);
+
+    pthread_cond_destroy(&awa.condition);
+    pthread_mutex_destroy(&awa.cond_lock);
+    return stanza;
+}

src/XMPPThread/internal.h

@@ -0,0 +1,77 @@
+#include <XMPPCommand.h>
+#include <Parsee.h>
+#include <XMPP.h>
+#include <XML.h>
+
+#define IQ_ADVERT \
+            AdvertiseSimple("http://jabber.org/protocol/chatstates") \
+            AdvertiseSimple("http://jabber.org/protocol/commands") \
+            AdvertiseSimple("http://jabber.org/protocol/caps") \
+            AdvertiseSimple("urn:xmpp:avatar:metadata+notify") \
+            AdvertiseSimple("urn:xmpp:avatar:data+notify") \
+            AdvertiseSimple("urn:xmpp:avatar:metadata") \
+            AdvertiseSimple("urn:xmpp:message-correct:0") \
+            AdvertiseSimple("urn:xmpp:message-moderate:0") \
+            AdvertiseSimple("urn:xmpp:message-moderate:1") \
+            AdvertiseSimple("urn:xmpp:message-retract:0") \
+            AdvertiseSimple("urn:xmpp:message-retract:1") \
+            AdvertiseSimple("urn:xmpp:avatar:data") \
+            AdvertiseSimple("urn:xmpp:chat-markers:0") \
+            AdvertiseSimple("urn:xmpp:reactions:0") \
+            AdvertiseSimple("urn:xmpp:styling:0") \
+            AdvertiseSimple("urn:xmpp:receipts") \
+            AdvertiseSimple("urn:xmpp:reply:0") \
+            AdvertiseSimple("jabber:x:oob") \
+            AdvertiseSimple("vcard-temp") \
+            AdvertiseSimple("jabber:iq:version") \
+            AdvertiseSimple("urn:parsee:x-parsee:0") \
+            AdvertiseSimple("urn:parsee:jealousy:0")
+
+#define IQ_IDENTITY \
+            IdentitySimple("gateway", "matrix", "Parsee Matrix Gateway") \
+            IdentitySimple("client",  "pc", NAME " v" VERSION " bridge") \
+            IdentitySimple("component", "generic", "Parsee's component")
+
+typedef struct XMPPIdentity {
+    char *category, *type, *lang, *name;
+} XMPPIdentity;
+
+typedef struct XMPPThread XMPPThread;
+typedef struct XMPPThreadInfo {
+    /* A FIFO of stanzas inbound, to be read by dispatcher 
+     * threads. */
+    Array *stanzas;
+    pthread_mutex_t lock;
+
+    ParseeData *args;
+    XMPPComponent *jabber;
+    XMPPCommandManager *m;
+
+    struct XMPPThread *dispatchers;
+    size_t available_dispatchers;
+
+    bool running;
+    pthread_mutex_t chk_lock;
+} XMPPThreadInfo;
+struct XMPPThread {
+    pthread_t thr;
+    XMPPThreadInfo *info;
+};
+
+extern int ICollate(unsigned char *cata, unsigned char *catb);
+extern int IdentitySort(void *idap, void *idbp);
+
+extern char * ParseeGetBridgedRoom(ParseeData *data, XMLElement *stanza);
+extern char * ParseeGetEventFromID(ParseeData *data, XMLElement *stanza, char *id);
+extern char * ParseeGetReactedEvent(ParseeData *data, XMLElement *stanza);
+extern void ParseePushAllStanza(ParseeData *args, XMLElement *stanza, char *event);
+extern bool ParseeVerifyAllStanza(ParseeData *args, XMLElement *stanza);
+
+extern HashMap * ShoveStanza(HashMap *content, XMLElement *stanza);
+extern void ManageProfileItem(ParseeData *args, XMLElement *item, XMLElement *stanza, XMPPThread *thr);
+extern XMLElement * CreatePubsubRequest(char *from, char *to, char *node);
+
+
+extern bool MessageStanza(ParseeData *args, XMLElement *stanza, XMPPThread *thr);
+extern void IQStanza(ParseeData *args, XMLElement *stanza, XMPPThread *thr);
+extern void PresenceStanza(ParseeData *args, XMLElement *stanza);

src/include/Parsee.h

@@ -236,4 +236,8 @@ extern uint64_t ParseeUptime(void);
  * Returns: A human-readable string showing the duration[LA:HEAP]
  * Modifies: NOTHING */
 extern char * ParseeStringifyDate(uint64_t millis);
+
+/* Generates the JID of the Parsee bridge user. */
+extern char * ParseeJID(ParseeData *data);
+extern char * ParseeMXID(ParseeData *data);
 #endif

src/include/XMPP.h

@@ -25,7 +25,7 @@ extern XMPPComponent * XMPPInitialiseCompStream(char *host, int port);
 extern bool XMPPAuthenticateCompStream(XMPPComponent *comp, char *shared);
 
 /* Makes a user join/leave a MUC */
-extern void XMPPJoinMUC(XMPPComponent *comp, char *fr, char *muc);
+extern bool XMPPJoinMUC(XMPPComponent *comp, char *fr, char *muc);
 extern void XMPPLeaveMUC(XMPPComponent *comp, char *fr, char *muc, char *r);
 
 /* TODO: XMPP stuff, I don't fucking know, I'm not a Jabbernerd. */
@@ -96,4 +96,6 @@ extern char * XMPPGenerateVer(void);
 
 /* Annotates a presence with https://xmpp.org/extensions/xep-0115.html */
 extern void XMPPAnnotatePresence(XMLElement *presence);
+
+extern bool XMPPHasError(XMLElement *stanza, char *type);
 #endif