vnctalk-prosody/db-customization/prosody-trigger-noowner.sql

-- for processing inserts with trigger

create table if not exists processed_messages (
  sort_id bigint,
  body text,
  date integer,
  "from" text,
  "to" text,
  owner text,
  broadcast text,
  "broadcast-sender" text,
  room text,
  type text,
  id text,
  receipts text,
  x_attachment text,
  x_location text,
  x_replaceMsgId text,
  x_origMessage text,
  x_vncConference text,
  x_forwardMessage text,
  mention text,
  PRIMARY KEY (sort_id)
);


create index si2 on processed_messages ("from");
create index si3 on processed_messages ("to");
create index si4 on processed_messages (owner);
create index si5 on processed_messages (broadcast);
create index si6 on processed_messages (room);
create index si7 on processed_messages (type);
create index si8 on processed_messages (id);

alter table processed_messages add column htmlbody text;
alter table processed_messages add column topicid text;
alter table processed_messages add column parent0 text;
alter table processed_messages add column parent text;
alter table processed_messages add column topic text;
alter table processed_messages add column updated integer;
alter table processed_messages add column broadcast_title text;
alter table processed_messages add column group_action text;
alter table processed_messages add column "encrypted" text;
alter table processed_messages add column encryption text;
alter table processed_messages add column reactions text;
alter table processed_messages add column expiry integer;
alter table processed_messages add column starredBy text[];
create index si9 on processed_messages (expiry);
create index di1 on processed_messages (date);


create or replace rule upsert_processed_messages as
  on insert to processed_messages where (exists (select 1 from processed_messages where sort_id=NEW.sort_id))
  do instead
    update processed_messages set
      body = NEW.body,
      htmlbody = NEW.htmlbody,
      topicid = NEW.topicid,
      parent0 = NEW.parent0,
      parent = NEW.parent,
      topic = NEW.topic,
      date = NEW.date,
      "from" = NEW.from,
      "to" = NEW.to,
      owner = NEW.owner,
      broadcast = NEW.broadcast,
      "broadcast-sender" = NEW."broadcast-sender",
      room = NEW.room,
      type = NEW.type,
      id = NEW.id,
      receipts = NEW.receipts,
      x_attachment = NEW.x_attachment,
      x_location = NEW.x_location,
      x_replaceMsgId = NEW.x_replaceMsgId,
      x_origMessage = NEW.x_origMessage,
      x_vncConference = NEW.x_vncConference,
      group_action = NEW.group_action,
      mention = NEW.mention,
      expiry = NEW.expiry
    where sort_id=NEW.sort_id;

create table if not exists call_tracking (
  callid bigint,
  started_at integer,
  updated_at integer,
  caller text,
  conferenceId text,
  receipient text,
  state text,
  PRIMARY KEY (callid)
);

create index cri1 on call_tracking (receipient);
create index csi1 on call_tracking (started_at);

create or replace function fprocess_messages()
  returns trigger AS
  $BODY$
  DECLARE
    _body text;
    _htmlbody text;
    _topic text;
    _parent text;
    _parent0 text;
    _x_attachment text;
    _x_location text;
    _x_replaceMsgId text;
    _original_message text;
    _x_conference text;
    _x_forwardMessage text;
    _group_action text;
    _mention text;
    _x_encrypted text;
    _encryption text;
    _message_id text;
    _origtarget text;
    _from_bare text;
    _x_invite text;
    _x_broadcast text;
    _x_when integer;
    _from text;
    _to text;
    _received_id text;
    _broadcast_title text;
    _x_conference_scheduler text;
    _x_conferencekey text;
    _x_conference_from text;
    _x_role text;
    _x_affiliation text;

  BEGIN
    _body := (xpath('/message/body/text() | /message/jc:body/text() | /jc:message/jc:body/text()', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text;
    _htmlbody := (xpath('//message/html/* | //message/xim:html/* | //jc:message/xim:html/*', NEW.value::xml, ARRAY[ARRAY['xim', 'http://jabber.org/protocol/xhtml-im'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _topic := (xpath('/message/xc:topic/@name | /jc:message/xc:topic/@name', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _parent0 := (xpath('/message/xc:topic/@parent0 | /jc:message/xc:topic/@parent0', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _parent := (xpath('/message/xc:topic/@parent | /jc:message/xc:topic/@parent', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _x_attachment := (xpath('//attachment | //xc:attachment', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
    _x_location := (xpath('/message/xc:location | /jc:message/xc:location', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _x_replaceMsgId := (xpath('/message/mc0:replace/@id | /jc:message/mc0:replace/@id', NEW.value::xml, ARRAY[ARRAY['mc0', 'urn:xmpp:message-correct:0'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _original_message := (xpath('//originalMessage | //xc:originalMessage', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
    _x_conference := (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
    _x_forwardMessage := (xpath('/message/xc:forwardMessage | /jc:message/xc:forwardMessage', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _group_action := (xpath('//group_action/type/text() | //xc:group_action/xc:type/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
    _mention := (to_jsonb(xpath('/message/xr:reference/@uri | /jc:message/xr:reference/@uri', NEW.value::xml, ARRAY[ARRAY['xr', 'urn:xmpp:reference:0'], ARRAY['jc', 'jabber:client']])))::jsonb;
    _x_encrypted := (xpath('//jc:message/xo:encrypted |//message/xo:encrypted', NEW.value::xml , ARRAY[ARRAY['jc','jabber:client'],ARRAY['xo','urn:xmpp:omemo:1']]))[1]::text;
    _encryption := (xpath('//jc:message/xo:encryption |//message/xo:encryption', NEW.value::xml , ARRAY[ARRAY['jc','jabber:client'],ARRAY['xo','urn:xmpp:eme:0']]))[1]::text;
    _message_id := (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text;
    _broadcast_title := (xpath('//message/xc:vncTalkBroadcast/@title', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1];
    _origtarget := (xpath('//message/@origtarget | /message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text;
    _x_invite := (xpath('//message/nsx:x/nsx:invite/@from'::text, new.value::xml, ARRAY[ARRAY['nsx'::text, 'http://jabber.org/protocol/muc#user'::text]]))[1]::text;
    _x_broadcast := xpath('//message/xc:vncTalkBroadcast'::text, new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]);
    _x_when := extract(epoch from now())::integer;
    _from := (xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text;
    _from_bare := split_part(_from, '/', 1);
    _to := (xpath('/message/@to | /jc:message/@to', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text;
    _x_conference_scheduler := (xpath('/message/xc:vncTalkConferenceScheduler | /jc:message/xc:vncTalkConferenceScheduler', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text;
    _x_role := (xpath('//message/mu:x/mu:item/@role', NEW.value::xml, ARRAY[ARRAY['mu','http://jabber.org/protocol/muc#user']]))[1]::text;
    _x_affiliation := (xpath('//message/mu:x/mu:item/@affiliation', NEW.value::xml, ARRAY[ARRAY['mu','http://jabber.org/protocol/muc#user']]))[1]::text;

    if NEW.store = 'muc_log' then
        insert into processed_messages (sort_id, body, htmlbody, topic, parent0, parent, date, "from", "to", owner, room, type, id,
          x_attachment, x_location, x_replaceMsgId, x_origMessage, x_vncConference, x_forwardMessage, group_action, mention, "encrypted", encryption)
        select
          NEW.sort_id as sort_id,
          _body as body, _htmlbody as htmlbody, _topic as topic, _parent0 as parent0, _parent as parent,
          extract(epoch from now())::integer as date,
          room_nick_jid_map.user_jid as "from",
          NEW.user||'@'||NEW.host as "to",
          NEW.user||'@'||NEW.host as owner,
          NEW.user||'@'||NEW.host as room,
          'groupchat' as type,
          (case
            when _message_id IS NULL then NEW.key
            else _message_id
           end) as id,
          _x_attachment as x_attachment, _x_location as x_location, _x_replaceMsgId as x_replaceMsgId, _original_message as original_message,
          _x_conference as x_conference, _x_forwardMessage as x_forwardMessage, _group_action as group_action, _mention as mention,
          _x_encrypted as "encrypted", _encryption as encryption
        from room_nick_jid_map
        where (room_nick_jid_map.nickname = _from);

        if (_x_role = 'participant' and _x_affiliation = 'none') then
          insert into processed_messages (sort_id, body, htmlbody, topic, parent0, parent, date, "from", "to", owner, room, type, id,
            x_attachment, x_location, x_replaceMsgId, x_origMessage, x_vncConference, x_forwardMessage, group_action, mention, "encrypted", encryption)
          select
            NEW.sort_id as sort_id,
            _body as body, _htmlbody as htmlbody, _topic as topic, _parent0 as parent0, _parent as parent,
            extract(epoch from now())::integer as date,
            split_part(_from, '/', 2) as "from",
            NEW.user||'@'||NEW.host as "to",
            NEW.user||'@'||NEW.host as owner,
            NEW.user||'@'||NEW.host as room,
            'groupchat' as type,
            (case
              when _message_id IS NULL then NEW.key
              else _message_id
            end) as id,
            _x_attachment as x_attachment, _x_location as x_location, _x_replaceMsgId as x_replaceMsgId, _original_message as original_message,
            _x_conference as x_conference, _x_forwardMessage as x_forwardMessage, _group_action as group_action, _mention as mention,
            _x_encrypted as "encrypted", _encryption as encryption;

        end if;

        if (jsonb_array_length(_mention::jsonb) > 0) then
          insert into totalmentions (username, target, key, type) select
            split_part(split_part(jsonb_array_elements(_mention::jsonb)::text, 'xmpp:',2), '"',1) as username,
            NEW.user || '@' || NEW.host as target,
            NEW.key as key,
            'groupchat' as type;
        end if;


        if NEW.value like '%ENDED_CALL</type></group_action%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          _x_conference_from := (xpath('//xc:vncTalkConference/xc:from/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          update recent_history_table
          set has_active_call = false,
            updated_at=_x_when,
            last_callstate_update=_x_when
          where target=NEW.user||'@'||NEW.host;
          update call_tracking set
            updated_at = _x_when,
            state = 'ended'
          where callid in (select callid from call_tracking where receipient=_x_conference_from and conferenceid=_x_conferencekey order by callid desc limit 1);
        end if;
        if NEW.value like '%<eventType>leave</eventType>%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          _x_conference_from := (xpath('//xc:vncTalkConference/xc:from/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          update call_tracking set
            updated_at = _x_when,
            state = 'leave'
          where callid in (select callid from call_tracking where receipient=_x_conference_from and conferenceid=_x_conferencekey order by callid desc limit 1);
        end if;
        if NEW.value like '%<eventType>join</eventType>%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          _x_conference_from := (xpath('//xc:vncTalkConference/xc:from/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          update recent_history_table
            set has_active_call = true,
              updated_at=_x_when,
              last_callstate_update=_x_when
            where target=NEW.user||'@'||NEW.host;
          update call_tracking set
            updated_at = _x_when,
            state = 'join'
          where callid in (select callid from call_tracking where receipient=_x_conference_from and conferenceid=_x_conferencekey order by callid desc limit 1);
        end if;

        if _x_conference_scheduler is not null then
          insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, mentions, incoming, received_receipt, updated_at, deleted, has_data, x_conference_scheduler, x_conference_start) select
            distinct on (room_membership.username) room_membership.username as username,
            NEW.user || '@' || NEW.host as target,
            'groupchat' as type,
            extract(epoch from now()) as timestamp,
            _body as message,
            _message_id AS message_id,
            _original_message as original_message,
            _x_attachment as x_attachment, _x_conference as x_conference, _mention as mentions,
            (case
              when (_from = room_membership.user_room_nickname) then false
              else true
            end) as incoming,
            false as received_receipt,
            extract(epoch from now()) as updated_at,
            ((xpath('//group_action/type/text() | //xc:group_action/xc:type/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] is not null) as deleted,
            true as has_data,
            _x_conference_scheduler as x_conference_scheduler,
            extract(epoch from to_timestamp(((xpath('/message/xc:vncTalkConferenceScheduler/xc:startTime/text() | /jc:message/xc:vncTalkConferenceScheduler/xc:startTime/text()', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text), 'YYYY-MM-DD HH24:MI:SS')) as x_conference_start
            from room_membership where NEW.host=room_membership.host and NEW.user=room_membership.room
          order by username, room_membership.user_room_nickname desc;
        end if;

        if (_body is not null) and (_body <> '') then
          insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, mentions, incoming, received_receipt, updated_at, deleted, has_data) select
            distinct on (room_membership.username) room_membership.username as username,
            NEW.user || '@' || NEW.host as target,
            'groupchat' as type,
            extract(epoch from now()) as timestamp,
            _body as message, _message_id as message_id, _original_message as original_message, _x_attachment as x_attachment, _x_conference as x_conference,
            _mention as mentions,
            (case
              when (_from = room_membership.user_room_nickname) then false
              else true
             end) as incoming,
            false as received_receipt,
            _x_when as updated_at,
            (_group_action is not null) as deleted,
            true as has_data
            from room_membership where NEW.host=room_membership.host and NEW.user=room_membership.room
            order by username, room_membership.user_room_nickname desc;


          if (_x_conference is null) then
            insert into read_conversation (username, target, timestamp)  select
              room_nick_jid_map.user_jid AS username,
              (new."user" || '@'::text) || new.host AS target,
               _x_when AS "timestamp"
              FROM room_nick_jid_map
              WHERE room_nick_jid_map.room_name = ((new."user" || '@'::text) || new.host) AND room_nick_jid_map.nickname = _from;
          end if;
        end if;
    end if;

    if NEW.store = 'muc_remote' then
      insert into processed_messages (sort_id, body, htmlbody, topic, parent0, parent, date, "from", "to", owner, room, type, id,
        x_attachment, x_location, x_replaceMsgId, x_origMessage, x_vncConference, x_forwardMessage, group_action, mention, "encrypted", encryption)
      select
        NEW.sort_id as sort_id,
        _body as body, _htmlbody as htmlbody, _topic as topic, _parent0 as parent0, _parent as parent,
        extract(epoch from now())::integer as date,
        room_nick_jid_map.user_jid as "from",
        NEW.user||'@'||NEW.host as "to",
        NEW.user||'@'||NEW.host as owner,
        NEW.with as room,
        'groupchat' as type,
        (case
          when _message_id IS NULL then NEW.key
          else _message_id
         end) as id,
         _x_attachment as x_attachment, _x_location as x_location, _x_replaceMsgId as _x_replaceMsgId, _original_message as original_message,
         _x_conference as x_conference, _x_forwardMessage as x_forwardMessage, _group_action as group_action, _mention as mention,
         _x_encrypted as "encrypted", _encryption as encryption
      from room_nick_jid_map
      where (room_nick_jid_map.nickname = _from);

      -- mentions
      if (jsonb_array_length(_mention::jsonb) > 0) then
        insert into totalmentions (username, target, key, type) select
          split_part(split_part(jsonb_array_elements(_mention::jsonb)::text, 'xmpp:',2), '"',1) as username,
          NEW.user || '@' || NEW.host as target,
          NEW.key as key,
          'groupchat' as type;
      end if;
      if (_group_action = 'UPDATE_ENCRYPTION') then
        if ((xpath('//group_action/data/text() | //xc:group_action/xc:data/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1]::text = 1) then
          update recent_history_table set direct_e2e=true where target=NEW.with;
        else
          update recent_history_table set direct_e2e=false where target=NEW.with;
        end if;
      end if;

      if NEW.value like '%ENDED_CALL</type></group_action%' then
        update recent_history_table
        set has_active_call = false,
          updated_at=extract(epoch from now())::integer,
          last_callstate_update=extract(epoch from now())::integer
        where target=NEW.with;
      end if;
      if NEW.value like '%<eventType>join</eventType>%' then
      update recent_history_table
        set has_active_call = true,
          updated_at=extract(epoch from now())::integer,
          last_callstate_update=extract(epoch from now())::integer
        where target=NEW.with;
      end if;

      if _x_conference_scheduler is not null then
        update recent_history_table
        set x_conference_scheduler=_x_conference_scheduler,
          x_conference_start=extract(epoch from to_timestamp(((xpath('/message/xc:vncTalkConferenceScheduler/xc:startTime/text() | /jc:message/xc:vncTalkConferenceScheduler/xc:startTime/text()', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text), 'YYYY-MM-DD HH:MI:SS'))
        where recent_history_table.target=NEW.with;
      end if;

      if (_body is not null) and (_body <> '') then
        insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, mentions, incoming, received_receipt, updated_at, deleted, has_data) select
          NEW.user||'@'||NEW.host as username,
          NEW.with as target,
          'groupchat' as type,
          extract(epoch from now()) as timestamp,
          _body as message, _message_id as message_id, _original_message as original_message, _x_attachment as x_attachment, _x_conference as x_conference,
          _mention as mentions,
          (case
            when (NEW.user||'@'||NEW.host = room_nick_jid_map.user_jid) then false
            else true
           end) as incoming,
          false as received_receipt,
          _x_when as updated_at,
          (_group_action is not null) as deleted,
          true as has_data
        from room_nick_jid_map
        where (room_nick_jid_map.nickname = _from);
      end if;

    end if;

    if NEW.store = 'archive'  then
      -- processed_messages
      if (NEW.with != NEW.user||'@'||NEW.host) then
        insert into processed_messages (sort_id, body, htmlbody, date, "from", "to", "broadcast-sender", owner, type, id, broadcast,
          x_attachment, x_location, x_replaceMsgId, x_origMessage, x_vncConference, x_forwardMessage, broadcast_title, group_action, "encrypted", encryption)
        select
          NEW.sort_id as sort_id,
          _body as body, _htmlbody as htmlbody,
          extract(epoch from now())::integer as date,
          (case
            when (_origtarget IS NULL) then  _from_bare
            else _origtarget
          end) as from,
          _to AS "to",
          (case
            when _origtarget IS NULL then
              NULL
            else _from_bare
          end) as "broadcast-sender",
          NEW.user||'@'||NEW.host as owner,
          'chat' as type,
          (case
            when _message_id IS NULL then NEW.key
            else _message_id
           end) as id,
           _origtarget as broadcast,
           _x_attachment as x_attachment, _x_location as x_location, _x_replaceMsgId as x_replaceMsgId, _original_message as original_message,
           _x_conference as x_conference, _x_forwardMessage as x_forwardMessage,
           _broadcast_title as broadcast_title,
           _group_action as group_action, _x_encrypted as "encrypted", _encryption as encryption
          where
            _x_invite IS NULL
          ;


        if NEW.value like '%<eventType>invite</eventType>%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          insert into conferencemap (conferencekey, value)
          select
            _x_conferencekey as conferencekey,
            '{"value":"'||(xpath('//xc:vncTalkConference/xc:jitsiRoom/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1]||'","jitsiurl":"'||(xpath('//xc:vncTalkConference/xc:jitsiURL/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1]||'"}' as value
          ;
          if NEW.with = _to then
            insert into call_tracking (callid, started_at, caller, conferenceId, receipient, state)
            select
              NEW.sort_id as callid,
              _x_when as started_at,
              _from_bare as caller,
              _x_conferencekey as conferenceId,
              _to as receipient,
              'invite' as state
            ;
          end if;
        end if;

        if NEW.value like '%<eventType>leave</eventType>%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          update call_tracking set
            updated_at = _x_when,
            state = 'leave'
          where callid in (select callid from call_tracking where (receipient=_from_bare or caller=_from_bare) and conferenceid=_x_conferencekey order by callid desc limit 1);
        end if;

        if NEW.value like '%<eventType>join</eventType>%' then
          _x_conferencekey := (xpath('//xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1];
          update call_tracking set
            updated_at = _x_when,
            state = 'join'
          where callid in (select callid from call_tracking where receipient=_from_bare and conferenceid=_x_conferencekey order by callid desc limit 1);
        end if;


        if (NEW.with like '%@conference.%') then
          -- init read conversation
          insert into read_conversation (username, target, timestamp)
          select
            (split_part(_to, '/',1)) as username,
            (split_part(_from, '/',1)) as target,
            (extract(epoch from now())::integer - 60) as timestamp;
          -- update room_nick_jid_map_inviter
          insert into room_nick_jid_map (room_name, user_jid, nickname, since)
            select
            (split_part(_from, '/',1)) as room_name,
            split_part(split_part(xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:x:conference']])::text, '/', 1), '"', 2) as user_jid,
            ((split_part(_from, '/',1))||'/'||split_part(split_part(xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:x:conference']])::text, '/', 1), '"', 2)) as nickname,
            extract(epoch from now())::integer;
          -- update room_nick_jid_map_invitee
          insert into room_nick_jid_map (room_name, user_jid, nickname, since)
          select
            (split_part((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1)) as room_name,
            (split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1)) as user_jid,
            ((split_part((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1))||'/'||(split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1))) as nickname,
            extract(epoch from now())::integer as since;

        else

          if (_body is not null) and ((_x_conference is null) or (_x_conference like '%#%')) and (_x_invite is null) then
            if (_origtarget is null) then
              -- update chat recent
              insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, incoming, received_receipt, updated_at, has_data, deleted) select
                NEW.user || '@' || NEW.host as username,
                NEW.with as target,
                'chat' as type,
                _x_when as timestamp,
                _body as message, _message_id as message_id, _original_message as original_message, _x_attachment as x_attachment, _x_conference as x_conference,
                (case
                  when (split_part(_to, '/',1) = NEW.user || '@' || NEW.host) then true
                  else false
                  end) as incoming,
                NULL as received_receipt,
                _x_when as updated_at,
                true as has_data,
                false as deleted
                order by target,timestamp desc;
            end if;
            if (_origtarget is not null) and (_x_broadcast is not null) then
              insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, broadcast_title, incoming, received_receipt, updated_at, has_data, deleted) select
                NEW.user || '@' || NEW.host as username,
                _origtarget as target, 'broadcast' as type, _x_when as timestamp,
                _body as message, _message_id as message_id, _original_message as original_message, _x_attachment as x_attachment, _x_conference as x_conference,
                _broadcast_title as broadcast_title,
                (case
                  when (split_part(_to, '/',1) = NEW.user || '@' || NEW.host) then true
                  else false
                  end) as incoming,
                NULL as received_receipt,
                _x_when as updated_at,
                true as has_data,
                false as deleted
                order by timestamp desc;
            end if;

            if (_origtarget is not null) and (_x_broadcast is not null) then
              insert into broadcast_audience (broadcast_owner, broadcast_target, audience, title) select
                NEW.with as broadcast_owner,
                _origtarget as broadcast_target,
                array_to_json(xpath('//message/xc:vncTalkBroadcast/xc:*/text()'::text, NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))::jsonb as audience,
                _broadcast_title as title;
            end if;

          end if;
        end if;
      end if;

      if (_to = NEW.with) and (_body is not null) and (_body <> '') then
        insert into read_conversation (username, target, "timestamp") select
          (new."user" || '@'::text) || new.host AS username,
          NEW.with as target,
          _x_when as timestamp;
      end if;


    end if;

    if NEW.store = 'kick' then
      delete from room_nick_jid_map where (room_name=NEW.user||'@'||NEW.host) and (user_jid=NEW.with);
      update read_conversation set timestamp=_x_when + 30 where username=NEW.with and (target=NEW.user||'@'||NEW.host);
      update recent_history_table set deleted = true where ((target=NEW.user||'@'||NEW.host) and (username = NEW.with));
    end if;

    if NEW.store = 'receipts' then
      _received_id := (xpath('//received/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text;

      update processed_messages set receipts = _x_when::text, updated = _x_when
      where (processed_messages.id = _received_id)
      and (processed_messages.owner = NEW.user||'@'||NEW.host) ;

      update recent_history_table set received_receipt=true, updated_at = _x_when
      where (recent_history_table.message_id = _received_id)
      and (recent_history_table.username = NEW.user||'@'||NEW.host);

    end if;

    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger process_messages after insert on prosodyarchive
  for each row execute procedure fprocess_messages();


-- create or replace function fupdate_room_nick_jid_old()
create or replace function fupdate_room_nick_jid()
  returns trigger AS
  $BODY$
  BEGIN
     delete from room_nick_jid_map where (room_name=NEW.user||'@'||NEW.host) and (user_jid not in (select jsonb_object_keys(NEW.value::jsonb) as user_jid));
     update recent_history_table set deleted = true where ((target=NEW.user||'@'||NEW.host) and username not in (select jsonb_object_keys(NEW.value::jsonb)));
     insert into room_nick_jid_map (room_name, user_jid, nickname, since) select
        NEW.user||'@'||NEW.host as room_name,
        jsonb_object_keys(NEW.value::jsonb) as user_jid,
        NEW.user||'@'||NEW.host||'/'||jsonb_object_keys(NEW.value::jsonb) as nickname,
        (extract(epoch from now())::integer - 60) as since;
     RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;


create or replace function fupdate_room_nick_jid()
  returns trigger AS
  $BODY$
  BEGIN
     delete from room_nick_jid_map where (room_name=NEW.user||'@'||NEW.host) and (user_jid not in (select jsonb_object_keys(NEW.value::jsonb) as user_jid));
     update recent_history_table set deleted = true where ((target=NEW.user||'@'||NEW.host) and username not in (select jsonb_object_keys(NEW.value::jsonb)));
     insert into room_nick_jid_map (room_name, user_jid, nickname, since) select
        NEW.user||'@'||NEW.host as room_name,
        jsonb_object_keys(NEW.value::jsonb) as user_jid,
        NEW.user||'@'||NEW.host||'/'||jsonb_object_keys(NEW.value::jsonb) as nickname,
        (extract(epoch from now())::integer - 60) as since;
     RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;


create trigger update_room_nick_jid after insert on prosody
 for each row
 when (NEW.key='_affiliations')
 execute procedure fupdate_room_nick_jid();

create or replace function fupdate_mention_stamp()
 returns trigger AS
 $BODY$
 BEGIN
    update recent_history_table set last_mention_time=extract(epoch from now())::integer
      where username=NEW.username and target=NEW.target;
    RETURN NEW;
 END;
 $BODY$
LANGUAGE plpgsql VOLATILE;

create or replace function fupdate_room_nick_jid_remote()
  returns trigger AS
  $BODY$
  BEGIN
    insert into room_nick_jid_map (room_name, user_jid, nickname, since) select
      split_part(NEW.key, '/',1) as room_name,
      NEW.value as user_jid,
      NEW.key as nickname,
      (extract(epoch from now())::integer - 60) as since
    where (NEW.key ilike '%/%');
    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger update_room_nick_jid_remote after insert on prosody
  for each row
  when (NEW.store='muc_remote' and (NEW.key ilike '%/%'))
  execute procedure fupdate_room_nick_jid_remote();


create trigger update_mention_stamp after insert on totalmentions
  for each row execute procedure fupdate_mention_stamp();


create or replace function fupdate_recent_avatar()
  returns trigger AS
  $BODY$
  BEGIN
    update recent_history_table
    set
      last_avatar_update = NEW.value::integer,
      updated_at=EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.target=NEW.user||'@'||new.host and NEW.store='avatarupdate';
    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger update_recent_avatar after insert on prosody
  for each row
  when (NEW.store='avatarupdate')
  execute procedure fupdate_recent_avatar();


create or replace function fupdate_iom_callstates()
  returns trigger as
  $BODY$
  BEGIN
    update recent_history_table
      set has_active_call=true,
      last_callstate_update = EXTRACT(EPOCH FROM now())::integer,
      updated_at = EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.target = NEW.user and NEW.value='callactive';
    update recent_history_table
      set has_active_call=false,
      last_callstate_update = EXTRACT(EPOCH FROM now())::integer,
      updated_at = EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.target = NEW.user and NEW.value='callinactive';
    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger update_iom_callstates after insert on prosody
  for each row
  when (NEW.store='callactive')
  execute procedure fupdate_iom_callstates();


create or replace function fupdate_msg_reactions()
  returns trigger AS
  $BODY$
  BEGIN
    update processed_messages
    set updated = EXTRACT(EPOCH FROM now())::integer,
    reactions = r.v from (select jsonb_agg(jsonb_build_object(reactor, reaction)) as v from emojireactions where emojireactions.msgid=NEW.msgid) as r
    where processed_messages.id=NEW.msgid;
    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger update_msg_reactions after insert or update on emojireactions
for each row execute procedure fupdate_msg_reactions();


create or replace function fupdate_msg_delreactions()
  returns trigger AS
  $BODY$
  BEGIN
    update processed_messages
    set updated = EXTRACT(EPOCH FROM now())::integer,
    reactions = r.v from (select jsonb_agg(jsonb_build_object(reactor, reaction)) as v from emojireactions where emojireactions.msgid=OLD.msgid) as r
    where processed_messages.id=OLD.msgid;
    RETURN OLD;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;


create trigger update_msg_delreactions after delete on emojireactions
for each row execute procedure fupdate_msg_delreactions();


CREATE TABLE if not exists groupchat (
    roomjid text NOT NULL,
    roomtitle text,
    PRIMARY KEY (roomjid)
);

CREATE or replace RULE upsert_groupchat AS
  ON INSERT TO groupchat  WHERE (EXISTS ( SELECT 1 FROM groupchat WHERE (roomjid = new.roomjid)))
  DO INSTEAD
    UPDATE groupchat SET roomtitle = new.roomtitle
  WHERE (groupchat.roomjid = new.roomjid);


CREATE or replace FUNCTION fupdate_groupchat()
  RETURNS trigger AS
  $BODY$
  BEGIN
    if NEW.store = 'config' AND NEW.key='_data' then
      insert into groupchat (roomjid, roomtitle) select
        NEW.user||'@'||NEW.host as roomjid,
        NEW.value::jsonb->>'subject' as roomtitle;
    end if;
    RETURN NEW;
  END;
  $BODY$
LANGUAGE plpgsql VOLATILE;

create trigger update_groupchat AFTER INSERT ON prosody
FOR EACH ROW WHEN (new.key = '_data'::text AND new.store = 'config'::text)
EXECUTE PROCEDURE fupdate_groupchat();