Parsee/src/XMPPThread.c

#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 <Matrix.h>
#include <XMPP.h>
#include <XML.h>
#include <AS.h>

/* TODO: Rewrite this avatar code.
 * XEP-0084 sucks. */
#define IQ_ADVERT \
            AdvertiseSimple("http://jabber.org/protocol/chatstates") \
            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: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")

/* TODO: More identities */
#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;

static char *
ParseeJID(ParseeData *data)
{
    return StrConcat(2, "parsee@", data->config->component_host);
}

static 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;
}

static 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);
}

/* 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;
}

static 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;
}
static 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;
}
static 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);
}

static 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);
}

static 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;
}
struct XMPPThread;
typedef struct XMPPThreadInfo {
    /* A FIFO of stanzas */
    Array *stanzas;
    pthread_mutex_t lock;

    ParseeData *args;
    XMPPComponent *jabber;

    struct XMPPThread *dispatchers;
    size_t available_dispatchers;

    bool running;
    pthread_mutex_t chk_lock;
} XMPPThreadInfo;
typedef struct XMPPThread {
    pthread_t thr;
    XMPPThreadInfo *info;
} XMPPThread;

/* 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;
}
static 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;
}
static 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);
}

static 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;
    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;
    }

    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
    body = XMLookForUnique(stanza, "body");
    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: 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;
        }
    }

    mroom_id = ParseeGetBridgedRoom(args, stanza);
    if (mroom_id && !XMPPIsParseeStanza(stanza))
    {
        char *res = ParseeGetResource(from);
        char *encoded   = ParseeEncodeJID(args->config, decode_from, false);
        char *event_id  = NULL;
        char *replaced  = XMPPGetReplacedID(stanza);
        char *retracted = XMPPGetRetractedID(stanza);
        char *reply_to  = XMPPGetReply(stanza);
        bool chat       = false;

        if (StrEquals(HashMapGet(stanza->attrs, "type"), "chat"))
        {
            Free(encoded);
            encoded = StrDuplicate(from_matrix);
            chat = true;
        }

        {
            char *parsee = ParseeJID(args);

            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, "external_url", 
                        JsonValueString(oob_data->data)
                    );

                    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", 
                        MatrixCreateReact(event_id, react_data->data)
                    ));
                }
                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);
                }
                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", MatrixCreateReplace(event_id, data->data)
            ));
            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;
}

#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);

        /* TODO: Advertise more things */
        IQ_ADVERT
#undef AdvertiseSimple
    }

    return query;
}
static 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);
}
static 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;
}
static void
IQResult(ParseeData *args, XMLElement *stanza)
{
    XMLElement *vcard = XMLookForTKV(stanza, "vCard", "xmlns", "vcard-temp");

    XMLElement *event = XMLookForTKV(stanza, "pubsub", 
        "xmlns", "http://jabber.org/protocol/pubsub"
    );
    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");
        XMLElement *nickname = XMLookForUnique(vcard, "NICKNAME");

        if (nickname)
        {
            XMLElement *data = ArrayGet(nickname->children, 0);
            /* TODO: Use the nickname for something. 
             * And the rest of the vCard, somewhere. */
            (void) data;
        }
        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;
            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"));
            if (jid && !StrEquals(jid, HashMapGet(stanza->attrs, "from")))
            {
                from_matrix = ParseeEncodeJID(args->config, jid, false);
                ASSetAvatar(args->config, from_matrix, mxc);
            }
            else if (room)
            {
                char *mask = StrConcat(4, 
                    "@", args->config->sender_localpart, 
                    ":", args->config->homeserver_host
                );
                HashMap *obj = HashMapCreate();

                HashMapSet(obj, "url", JsonValueString(mxc));
                ASSetState(
                    args->config, 
                    room, "m.room.avatar", "", 
                    mask, obj
                );
                Free(mask);
            }
            
            Free(from_matrix);
            Free(room);
            Free(jid);
            Free(mxc);
        }
    }
}
static void
IQGet(ParseeData *args, XMLElement *stanza)
{
    XMPPComponent *jabber = args->jabber;
    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;
        char *from = HashMapGet(stanza->attrs, "from");
        char *to = HashMapGet(stanza->attrs, "to");
        char *id = HashMapGet(stanza->attrs, "id");

        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());
    }

}
static void
IQError(ParseeData *args, XMLElement *stanza)
{
    /* TODO */
}
#undef DISCO
static void
IQStanza(ParseeData *args, XMLElement *stanza)
{
    char *type;
    type = HashMapGet(stanza->attrs, "type");
#define OnType(ctyp, callback)  do \
                                { \
                                    if (StrEquals(type, #ctyp)) \
                                    { \
                                        callback(args, stanza); \
                                        return; \
                                    } \
                                } \
                                while (0)
    
    OnType(get, IQGet);
    OnType(error, IQError);
    OnType(result, IQResult);
#undef OnType
}

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 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");
    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 *from = NULL;
        char *type = HashMapGet(stanza->attrs, "type");
        char *room = ParseeGetBridgedRoom(args, stanza);
        char *decode_from = ParseeLookupJID(oid);
        char *from_matrix = ParseeDecodeMXID(decode_from);
        char *affiliation = HashMapGet(item->attrs, "affiliation");
        int power_level = 0;

        if (!from_matrix || *from_matrix != '@')
        {
            Free(from_matrix);
            from_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"))
        {
            power_level = 0;
        }
        else if (StrEquals(affiliation, "outcast"))
        {
            power_level = -1;
        }

        /* Set the user's PL */
        if (room)
        {
            HashMap *powers = ASGetPL(args->config, room);
            int64_t level = GrabInteger(powers, 2, "users", from_matrix);

            if (powers && level != power_level)
            {
                HashMap *users = GrabObject(powers, 1, "users");
                JsonValue *val = JsonValueInteger(power_level);
                JsonValueFree(HashMapSet(users, 
                    from_matrix, val
                ));
                ASSetPL(args->config, room, powers);
            }
            else
            {
                JsonFree(powers);
            }
        }

        if (StrEquals(type, "unavailable"))
        {
            /* If not an MXID, use the Parsee user */
            if (IsStatus(301))
            {
                ASBan(args->config, room, from_matrix);
            }
            else if (IsStatus(307))
            {
                ASKick(args->config, room, from_matrix);
            }
            else if (IsStatus(303))
            {
                char *nick = HashMapGet(item->attrs, "nick");
                ASSetName(args->config, from_matrix, nick);
            }
            else
            {
                ASLeave(args->config, room, from_matrix);
            }

        }

        if (jid)
        {
            ParseePushJIDTable(oid, jid);
        }

        Free(from);
        Free(decode_from);
        Free(from_matrix);
        Free(room);
    }
    else 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 */
    if (0)
    {
        XMPPComponent *jabber = args->jabber;
        char *from = ParseeJID(args);

        XMLElement *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);
    }
#undef MUC_USER_NS
}

static 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);
 
        if (!stanza)
        {
            UtilSleepMillis(1);
            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);
        }
        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;

    XMLElement *stanza;
} XMPPAwait;
static pthread_mutex_t await_lock = PTHREAD_MUTEX_INITIALIZER;
static HashMap *await_table = NULL;

void *
ParseeXMPPThread(void *argp)
{
    ParseeData *args = argp;
    XMPPComponent *jabber = args->jabber;
    XMLElement *stanza = NULL;
    XMPPThreadInfo info;
    size_t i;
        
    /* Initialise the await table */
    await_table = HashMapCreate();

    /* Initialise the FIFO */
    info.stanzas = ArrayCreate();
    pthread_mutex_init(&info.lock, NULL);

    /* TODO: Make that configurable. */
    info.available_dispatchers = 8;
    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)))
            {
                await->stanza = stanza;             
                pthread_mutex_lock(&await->cond_lock);
                pthread_cond_signal(&await->condition);
                pthread_mutex_unlock(&await->cond_lock);

                HashMapDelete(await_table, id);

                pthread_mutex_unlock(&await_lock);
                continue;
            }
            pthread_mutex_unlock(&await_lock);
        }

        PushStanza(&info, stanza);
    }

    info.running = false;
    for (i = 0; i < info.available_dispatchers; i++)
    {
        pthread_t thr = info.dispatchers[i].thr;
        pthread_join(thr, NULL);
    }
    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);
    return NULL;
}
XMLElement *
ParseeAwaitStanza(char *identifier)
{
    /* 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;
    if (!identifier)
    {
        return NULL;
    }

    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)
    {
        /* TODO: Allow timeouts. */
        pthread_cond_wait(&awa.condition, &awa.cond_lock);
    }

    stanza = awa.stanza;
    pthread_mutex_lock(&await_lock);
    pthread_mutex_unlock(&await_lock);

    pthread_mutex_unlock(&awa.cond_lock);

    pthread_cond_destroy(&awa.condition);
    pthread_mutex_destroy(&awa.cond_lock);
    return stanza;
}