vnctalk-prosody/db-customization/prosody-13-new-deployment.sql

-- =============================================================================
-- vnctalk-prosody 13.0.6 — full DB customization for NEW deployments
-- =============================================================================
-- This is the single idempotent script for fresh Prosody 13.0.6 deployments.
-- It REPLACES the two legacy 0.11.6 scripts:
--   - prosody-queries-noowner.sql.notifyfix
--   - prosody-trigger-noowner.sql
--
-- Differences from the legacy scripts:
--   * MUC rules/views/triggers adapted to the 13.0.6 storage layout
--     (affiliations stored one row per bare JID in the `prosody` `config`
--      store; `_affiliations` / `_occupants` keys no longer exist).
--   * `room_nicknames` view and `update_room_nick_jid_map` rule removed
--     (occupant data is no longer in the DB during normal operation;
--      `room_nick_jid_map` is derived purely from affiliations — all allowed
--      clients use their bare JID as their room nickname).
--   * `update_group_owners` rule replaced by a trigger (the legacy rule
--     conflicted with the existing `upsert_group_owners` INSTEAD rule).
--   * All `ALTER TABLE … ADD COLUMN` made idempotent (`IF NOT EXISTS`).
--   * One-time backfill INSERT/UPDATE statements removed (they matched
--     nothing on an empty DB and referenced the obsolete `_affiliations` key).
--
-- Idempotent: safe to re-run. Every table uses `IF NOT EXISTS`, every rule /
-- view / function uses `CREATE OR REPLACE`, and triggers are dropped before
-- re-creation. `ALTER TABLE … ADD COLUMN IF NOT EXISTS` is a no-op if the
-- column already exists.
--
-- Prerequisite: the Prosody storage tables `prosody` and `prosodyarchive`
-- must already exist (created by mod_storage_sql when Prosody first starts
-- against this DB). This script aborts if they are missing.
-- =============================================================================

\set ON_ERROR_STOP on

do $$
begin
  if not exists (select 1 from information_schema.tables where table_name = 'prosody') then
    raise exception 'Table "prosody" does not exist. Start Prosody 13.0.6 once so mod_storage_sql creates the storage tables, then re-run this script.';
  end if;
  if not exists (select 1 from information_schema.tables where table_name = 'prosodyarchive') then
    raise exception 'Table "prosodyarchive" does not exist. Start Prosody 13.0.6 once so mod_storage_sql creates the storage tables, then re-run this script.';
  end if;
end $$;

create extension if not exists pgcrypto;

-- ---------------------------------------------------------------------------
-- prosody_unique_index: the unique index mod_storage_sql creates on the
-- `prosody` table (host, user, store, key). On a truly fresh DB Prosody
-- creates this itself via create_table(); re-creating it here with IF NOT
-- EXISTS is a harmless no-op in that case and fixes deployments where the
-- table was created by an older Prosody version without the unique index.
--
-- Without the unique index, mod_storage_sql falls back to SELECT-then-INSERT
-- instead of ON CONFLICT upsert, which can create duplicate rows under
-- concurrent writes (common in the fcmtoken map store). Deduplicate before
-- creating the index so it does not fail on pre-existing duplicates.
-- ---------------------------------------------------------------------------

-- Remove duplicate rows, keeping only the last-written row per
-- (host, user, store, key). ctid is the physical row identifier; the
-- highest ctid is the most recently inserted row. No-op if no duplicates.
delete from prosody
where ctid in (
  select ctid from (
    select ctid,
           row_number() over (
             partition by host, "user", store, key
             order by ctid desc
           ) as rn
    from prosody
  ) t
  where rn > 1
);

create unique index if not exists prosody_unique_index
  on prosody ("host", "user", "store", "key");

-- =============================================================================
-- TABLES
-- =============================================================================

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 if not exists si2 on processed_messages ("from");
create index if not exists si3 on processed_messages ("to");
create index if not exists si4 on processed_messages (owner);
create index if not exists si5 on processed_messages (broadcast);
create index if not exists si6 on processed_messages (room);
create index if not exists si7 on processed_messages (type);
create index if not exists si8 on processed_messages (id);
alter table processed_messages add column if not exists htmlbody text;
alter table processed_messages add column if not exists topicid text;
alter table processed_messages add column if not exists parent0 text;
alter table processed_messages add column if not exists parent text;
alter table processed_messages add column if not exists topic text;
alter table processed_messages add column if not exists updated integer;
alter table processed_messages add column if not exists broadcast_title text;
alter table processed_messages add column if not exists group_action text;
alter table processed_messages add column if not exists "encrypted" text;
alter table processed_messages add column if not exists encryption text;
alter table processed_messages add column if not exists reactions text;
alter table processed_messages add column if not exists expiry integer;
alter table processed_messages add column if not exists starredBy text[];
create index if not exists si9 on processed_messages (expiry);
create index if not exists di1 on processed_messages (date);

create table if not exists recent_history_table (
  username text,
  target text,
  type text,
  timestamp integer,
  message text,
  original_message text,
  x_attachment text,
  x_conference text,
  incoming boolean,
  PRIMARY KEY (target, username)
);
alter table recent_history_table add column if not exists message_id text;
alter table recent_history_table add column if not exists received_receipt boolean;
alter table recent_history_table add column if not exists updated_at integer;
alter table recent_history_table add column if not exists mute_sound integer;
alter table recent_history_table add column if not exists mute_notification integer;
alter table recent_history_table add column if not exists has_data boolean;
alter table recent_history_table add column if not exists broadcast_title text;
alter table recent_history_table add column if not exists mentions text;
alter table recent_history_table add column if not exists sort_id bigint;
alter table recent_history_table add column if not exists deleted boolean;
alter table recent_history_table add column if not exists pad_read integer;
alter table recent_history_table add column if not exists has_pads boolean;
alter table recent_history_table add column if not exists last_mention_time integer;
alter table recent_history_table add column if not exists last_avatar_update integer;
alter table recent_history_table add column if not exists direct_e2e boolean;
alter table recent_history_table add column if not exists has_active_call boolean;
alter table recent_history_table add column if not exists last_callstate_update integer;
alter table recent_history_table add column if not exists x_conference_start integer;
alter table recent_history_table add column if not exists retention_time integer;
alter table recent_history_table add column if not exists x_conference_scheduler text;
alter table recent_history_table add column if not exists is_favourite boolean;
alter table recent_history_table add column if not exists is_pinned boolean;
alter table recent_history_table add column if not exists audience_only boolean;
alter table recent_history_table add column if not exists has_iom boolean;
alter table recent_history_table add column if not exists pin_order integer;
update recent_history_table set updated_at=0 where updated_at is NULL;
update recent_history_table set mute_notification=0 where mute_notification is NULL;
update recent_history_table set mute_sound=0 where mute_sound is NULL;
update recent_history_table set has_data=true where has_data is NULL;

create table if not exists archive_inactive_table (
  username text,
  target text,
  timestamp integer,
  content text,
  PRIMARY KEY (target, username)
);

create table if not exists broadcast_audience (
  broadcast_owner text,
  broadcast_target text,
  audience jsonb,
  title text,
  PRIMARY KEY (broadcast_owner, broadcast_target)
);
alter table broadcast_audience add column if not exists tags jsonb;
alter table broadcast_audience add column if not exists description text;

create table if not exists group_owners (
  room text,
  owner text,
  primary key (room)
);
alter table group_owners add column if not exists created integer;
alter table group_owners add column if not exists updated integer;

create table if not exists group_avatarids (
  room text,
  avatarid text,
  primary key (room)
);

create table if not exists conferenceMapping (
  conferenceKey text,
  value text,
  primary key (conferenceKey)
);

create table if not exists conferenceMap (
  conferenceKey text,
  value text,
  primary key (conferenceKey)
);

create table if not exists exclude_from_history (
  username text,
  target text,
  timestamp integer,
  PRIMARY KEY (target, username)
);

create table if not exists read_conversation (
  username text,
  target text,
  timestamp integer,
  PRIMARY KEY (target, username)
);

create table if not exists read_pad (
  username text,
  target text,
  timestamp integer,
  PRIMARY KEY (target, username)
);

create table if not exists external_email_invites (
  username text,
  meeting text,
  email text,
  timestamp integer,
  PRIMARY KEY (username, meeting, email)
);

drop view if exists total_mentions cascade;
drop table if exists total_mentions cascade;
create table if not exists totalmentions (
  username text,
  target text,
  key text,
  type text,
  PRIMARY KEY (target, username, key)
);

create table if not exists room_nick_jid_map (
  room_name text,
  user_jid text,
  nickname text,
  primary key (room_name, user_jid, nickname)
);
alter table room_nick_jid_map add column if not exists since integer default 0;

create table if not exists unread_message_ids (
  message_id text,
  sender text,
  receipient text,
  timestamp integer,
  primary key ( message_id, sender, receipient)
);

create table if not exists mute_conversation (
  username text,
  target text,
  PRIMARY KEY (target, username)
);

create table if not exists mute_notification (
  username text,
  target text,
  type integer,
  PRIMARY KEY (target, username)
);

create table if not exists profile_update_queue (
  username text,
  timestamp integer,
  PRIMARY KEY (username)
);

create table if not exists unread_message_mention_ids (
  message_id text,
  sender text,
  receipient text,
  timestamp integer,
  primary key ( message_id, sender, receipient)
);

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 if not exists cri1 on call_tracking (receipient);
create index if not exists csi1 on call_tracking (started_at);

create table if not exists no_notify_before (
  username text,
  before integer,
  PRIMARY KEY (username)
);

create table if not exists ep_author_ids (
  username text,
  authorid text,
  primary key (username)
);

create table if not exists ep_conv_group (
  convid text,
  groupid text,
  primary key (convid)
);

create table if not exists ep_pad_names (
  padid text,
  groupid text,
  padname text,
  primary key (padid, groupid)
);

create table if not exists fcm_log (
  id SERIAL,
  msgid text,
  fromjid text,
  tojid text,
  tofcmtoken text,
  todevicetype text,
  timestamp integer,
  opstatus text,
  primary key(id)
);

create table if not exists remote_muc_names (
  username text,
  remotemuc text,
  displayname text,
  primary key (username, remotemuc)
);

create table if not exists recordings (
  convkey text,
  callid text,
  fileid text,
  timestamp integer,
  primary key (convkey, fileid, timestamp)
);

create table if not exists surveymap (
  id SERIAL,
  groupchat text,
  jid text,
  survey text,
  expiry integer,
  PRIMARY KEY (id, groupchat, jid, survey)
);
create index if not exists ri3 on surveymap(expiry);
alter table surveymap add column if not exists name text;
alter table surveymap add column if not exists description text;
alter table surveymap add column if not exists status text;
alter table surveymap add column if not exists settings text;

create table if not exists surveyresults (
  id SERIAL,
  surveyid integer,
  results text,
  jid text,
  PRIMARY KEY (id, jid)
);
create index if not exists ri2 on surveyresults(surveyid);
alter table surveyresults add column if not exists status text;

create table if not exists emojireactions (
  msgid text,
  reactor text,
  reaction text,
  PRIMARY KEY (msgid, reactor)
);

create table if not exists room_join_requests (
  requester text,
  room text,
  date integer,
  PRIMARY KEY (requester, room)
);

create table if not exists user_custom_images (
  id SERIAL,
  jid text,
  updated integer,
  type text,
  filename text,
  base64data text,
  primary key(jid, filename)
);
create index if not exists uci0 on user_custom_images(id);
create index if not exists uci1 on user_custom_images(jid);
create index if not exists uci2 on user_custom_images(updated);

create table if not exists groupchat (
    roomjid text NOT NULL,
    roomtitle text,
    PRIMARY KEY (roomjid)
);

create sequence if not exists broadcast_id_seq
   START WITH 1
   INCREMENT BY 1
   NO MINVALUE
   NO MAXVALUE
   CACHE 1;

-- =============================================================================
-- RULES — upsert / dedup rules on the derived tables
-- =============================================================================

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 or replace rule insert_into_archive_inactive as
  on insert to archive_inactive_table where (exists (select 1 from archive_inactive_table where username=NEW.username and target=NEW.target))
  do instead
    update archive_inactive_table set timestamp=NEW.timestamp,
    content=NEW.content
    where target=NEW.target and username=NEW.username;

create or replace rule update_recent_archive as
  on update to archive_inactive_table
  do update recent_history_table set updated_at=extract(epoch from now())::integer
  where target=NEW.target and username=NEW.username;

create or replace rule upsert_broadcast_audience as
  on insert to broadcast_audience where (exists (select 1 from broadcast_audience where broadcast_target=NEW.broadcast_target))
  do instead
    update broadcast_audience set
      audience=NEW.audience,
      title=NEW.title,
      tags=NEW.tags,
      description=NEW.description
    where broadcast_target=NEW.broadcast_target;

create or replace rule upsert_group_owners as
  on insert to group_owners where (exists (select 1 from group_owners where room=NEW.room))
  do instead
    update group_owners set
      owner=NEW.owner
    where room=NEW.room;

create or replace rule upsert_group_avatarids as
  on insert to group_avatarids where (exists (select 1 from group_avatarids where room=NEW.room))
  do instead
    update group_avatarids set
      avatarid=NEW.avatarid
    where room=NEW.room;

create or replace rule upsert_confmap as
  on insert to conferenceMapping where (exists (select 1 from conferenceMapping where conferenceKey=NEW.conferenceKey))
  do instead update conferenceMapping
    set value=NEW.value
    where conferenceKey=NEW.conferenceKey;

create or replace rule upsert_conferencemap as
  on insert to conferenceMap where (exists (select 1 from conferenceMap where conferenceKey=NEW.conferenceKey))
  do instead update conferenceMap
    set value=NEW.value
    where conferenceKey=NEW.conferenceKey;

create or replace rule insert_into_exclude_from_history as
  on insert to exclude_from_history where (exists (select 1 from exclude_from_history where username=NEW.username and target=NEW.target))
  do instead
    update exclude_from_history set timestamp=NEW.timestamp
    where  username=NEW.username and target=NEW.target;

create or replace rule insert_into_read_conversation as
  on insert to read_conversation where (exists (select 1 from read_conversation where username=NEW.username and target=NEW.target))
  do instead
    update read_conversation set timestamp=NEW.timestamp
    where target=NEW.target and username=NEW.username;

create or replace rule insert_into_read_pad as
  on insert to read_pad where (exists (select 1 from read_pad where username=NEW.username and target=NEW.target))
  do instead
    update read_pad set timestamp=NEW.timestamp
    where target=NEW.target and username=NEW.username;

create or replace rule inser_external_email_invites as
  on insert to external_email_invites where (exists (select 1 from external_email_invites where username=NEW.username and meeting=NEW.meeting and email=NEW.meeting))
  do instead nothing;

create or replace rule update_read_conversation as
  on update to read_conversation do
    update recent_history_table set
      updated_at=EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.username=NEW.username and recent_history_table.target=NEW.target;

create or replace rule update_read_pad as
  on update to read_pad do
    update recent_history_table set
      pad_read=EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.username=NEW.username and recent_history_table.target=NEW.target;

create or replace rule upsert_totalmentions as
  on insert to totalmentions where (exists (select 1 from totalmentions where username=NEW.username and target=NEW.target and key=NEW.key))
  do instead nothing;

create or replace rule
  insert_room_nick_jid_map as
  on insert to room_nick_jid_map where (exists ( select 1 from room_nick_jid_map where room_name=NEW.room_name and user_jid=NEW.user_jid and nickname=NEW.nickname))
  do instead nothing;

create or replace rule
  insert_mute_conversation as
  on insert to mute_conversation where (exists ( select 1 from mute_conversation where username=NEW.username and target=NEW.target))
  do instead nothing;

create or replace rule
  update_mute_conversation as
  on update to mute_conversation
  do update recent_history_table
    set updated_at=EXTRACT(EPOCH FROM now())::integer
    where recent_history_table.username=NEW.username and recent_history_table.target=NEW.target;

create or replace rule upsert_mute_notification as
  on insert to mute_notification where (exists ( select 1 from mute_notification where username=NEW.username and target=NEW.target))
  do instead update mute_notification
  set type=NEW.type where username=NEW.username and target=NEW.target;

create or replace rule update_mute_notification as
  on update to mute_notification do update recent_history_table
  set updated_at=EXTRACT(EPOCH FROM now())::integer
  where recent_history_table.username=NEW.username and recent_history_table.target=NEW.target;

create or replace rule insert_mute_notification as
  on insert to mute_notification do update recent_history_table
  set updated_at=EXTRACT(EPOCH FROM now())::integer
  where recent_history_table.username=NEW.username and recent_history_table.target=NEW.target;

create or replace rule upsert_profile_update_queue as
  on insert to profile_update_queue where (exists (select 1 from profile_update_queue where username=NEW.username))
  do instead update profile_update_queue set
    timestamp=NEW.timestamp
  where username=NEW.username;

create or replace rule
  upsert_unread_message_ids as
  on insert to unread_message_ids where (exists ( select 1 from unread_message_ids where message_id=NEW.message_id and sender=NEW.sender and receipient=NEW.receipient))
  do instead nothing;

create or replace rule
  upsert_unread_message_mention_ids as
  on insert to unread_message_mention_ids where (exists ( select 1 from unread_message_mention_ids where message_id=NEW.message_id and sender=NEW.sender and receipient=NEW.receipient))
  do instead nothing;

create or replace rule upsert_remote_muc_names as
  on insert to remote_muc_names where (exists (select 1 from remote_muc_names where username=NEW.username and remotemuc=NEW.remotemuc))
  do instead update remote_muc_names set
    displayname = NEW.displayname
  where username=NEW.username and remotemuc=NEW.remotemuc;

create or replace rule upsert_ep_author_ids as
  on insert to ep_author_ids where (exists (select 1 from ep_author_ids where username=NEW.username))
  do instead update ep_author_ids set
    authorid=NEW.authorid
  where username=NEW.username;

create or replace rule upsert_ep_conv_ids as
  on insert to ep_conv_group where (exists (select 1 from ep_conv_group where convid=NEW.convid))
  do instead update ep_conv_group set
    groupid=NEW.groupid
  where convid=NEW.convid;

create or replace rule upsert_ep_pad_names as
  on insert to ep_pad_names where (exists (select 1 from ep_pad_names where padid=NEW.padid and groupid=NEW.groupid))
  do instead update ep_pad_names set
    padname=NEW.padname
  where padid=NEW.padid and groupid=NEW.groupid;

create or replace rule upsert_no_notifiy as
  on insert to no_notify_before where (exists (select 1 from no_notify_before where username=NEW.username))
  do instead
    update no_notify_before set
      before=NEW.before
    where username=NEW.username;

create or replace rule upsert_survey_results as
    on insert to surveyresults where (exists (select 1 from surveyresults where surveyid=NEW.surveyid and jid=NEW.jid))
    do instead
    update surveyresults set
      results = NEW.results,
      status = NEW.status
    where surveyid = NEW.surveyid and jid=NEW.jid;

create or replace rule upsert_emojireactions as
  on insert to emojireactions where (exists (select 1 from emojireactions where msgid=NEW.msgid and reactor=NEW.reactor))
  do instead
  update emojireactions set
    reaction = NEW.reaction
  where msgid=NEW.msgid and reactor=NEW.reactor;

create or replace rule upsert_room_join_requests as
  on insert to room_join_requests where (exists (select 1 from room_join_requests where requester=NEW.requester and room=NEW.room))
  do instead
  update room_join_requests set
    date = extract(epoch from now())::integer
    where requester=NEW.requester and room=NEW.room;

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

-- =============================================================================
-- RULES — on `prosody` (kv store)
-- =============================================================================

-- ---------------------------------------------------------------------------
-- NOTE: the vcard_muc / vcard / muc_remote side-effects used to be conditional
-- `DO` rules on `prosody`. They have been converted to AFTER INSERT (/OR
-- UPDATE) triggers because PostgreSQL rejects `INSERT ... ON CONFLICT ...
-- DO UPDATE` on any table that carries a conditional (WHERE) DO/DO ALSO rule
-- or a non-NOTHING DO INSTEAD rule (error: "INSERT with ON CONFLICT clause
-- cannot be used with table that has INSERT or UPDATE rules"). Prosody 13's
-- mod_storage_sql uses ON CONFLICT upserts against the `prosody` kv table
-- whenever `prosody_unique_index` exists (created above), so the rules broke
-- every kv upsert. Triggers do not block ON CONFLICT. The derived-table
-- INSTEAD upsert rules (upsert_group_avatarids, upsert_profile_update_queue,
-- upsert_remote_muc_names, insert_room_nick_jid_map) are unaffected — they
-- live on the derived tables and keep rewriting the inserts emitted here.
-- ---------------------------------------------------------------------------

-- vcard_muc room avatar cache (vcard_muc store still uses key='')
create or replace function fcache_group_avatarids()
  returns trigger as $BODY$
begin
  insert into group_avatarids (room, avatarid)
  select
    NEW.user || '@' || NEW.host as room,
    encode(digest(decode(split_part(split_part(split_part(NEW.value, 'BINVAL', 2), '__array":["', 2), '"', 1), 'base64'), 'sha1'), 'hex') as avatarid;
  return NEW;
end;
$BODY$ language plpgsql volatile;

drop trigger if exists cache_group_avatarids on prosody;
create trigger cache_group_avatarids after insert on prosody
  for each row
  when (NEW.store = 'vcard_muc' and NEW.key = '' and NEW.type = 'json' and NEW.host like 'conference.%')
  execute procedure fcache_group_avatarids();

-- vcard profile-update queue. Was two rules (on insert + on update); merged
-- into one AFTER INSERT OR UPDATE trigger so the UPDATE branch of an ON
-- CONFLICT upsert is also covered (an AFTER INSERT trigger alone would miss
-- the conflict-update path, since ON CONFLICT DO UPDATE fires AFTER UPDATE
-- triggers, not AFTER INSERT, when the conflict is taken).
create or replace function fupdate_profile_queue()
  returns trigger as $BODY$
begin
  insert into profile_update_queue (username, timestamp)
  select
    NEW.user || '@' || NEW.host as username,
    extract(epoch from now())::integer as timestamp;
  return NEW;
end;
$BODY$ language plpgsql volatile;

drop trigger if exists update_profile_queue on prosody;
create trigger update_profile_queue after insert or update on prosody
  for each row
  when (NEW.store = 'vcard' and NEW.type = 'json' and NEW.key = '')
  execute procedure fupdate_profile_queue();

-- muc_remote nickname map (VNCtalk store)
create or replace function fupdate_muc_remote_name()
  returns trigger as $BODY$
begin
  insert into remote_muc_names (username, remotemuc, displayname)
  select
    NEW.user || '@' || NEW.host as username,
    NEW.key as remotemuc,
    NEW.value as displayname;
  return NEW;
end;
$BODY$ language plpgsql volatile;

drop trigger if exists update_muc_remote_name on prosody;
create trigger update_muc_remote_name after insert on prosody
  for each row
  when (NEW.store = 'muc_remote' and not (NEW.value ilike '%/%'))
  execute procedure fupdate_muc_remote_name();

create or replace function fupdate_room_nick_jid_map_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 as since;
  return NEW;
end;
$BODY$ language plpgsql volatile;

drop trigger if exists update_room_nick_jid_map_remote on prosody;
create trigger update_room_nick_jid_map_remote after insert on prosody
  for each row
  when (NEW.store = 'muc_remote' and NEW.key like '%/%' and NEW.value like '%@%')
  execute procedure fupdate_room_nick_jid_map_remote();

-- Drop any prior rule counterparts these triggers replace (no-op on a truly
-- fresh DB; cleans up a DB that previously ran an older script version).
drop rule if exists cache_group_avatarids on prosody;
drop rule if exists update_profile_queue_from_insert on prosody;
drop rule if exists update_profile_queue_from_update on prosody;
drop rule if exists update_muc_remote_name on prosody;
drop rule if exists update_room_nick_jid_map_remote on prosody;
-- Legacy 0.11.6 rule (fires on key='_affiliations', dead under 13.0.6).
-- PostgreSQL checks rule *existence* at plan time, so even this dead
-- conditional rule blocks ON CONFLICT — dropping it is mandatory.
drop rule if exists update_group_owners on prosody;

-- NOTE: group_owners is maintained by the `fupdate_group_owners` trigger fired
-- on the `_data` row (see trigger section below); the legacy
-- `update_group_owners` rule conflicted with the `upsert_group_owners`
-- INSTEAD rule and is dropped above.

-- =============================================================================
-- VIEWS (created early — several prosodyarchive rules below depend on them)
-- =============================================================================

-- room_membership: 13.0.6 layout — affiliations are one row per bare JID
-- (key=<jid>, value=<affiliation string>). Replaces the legacy
-- `where key='_affiliations'` + jsonb_object_keys form.
create or replace view room_membership as
  select
    prosody.key as username,
    ''::text as user_room_nickname,
    prosody.host as host,
    prosody.user as room
  from prosody
  where prosody.store = 'config'
    and prosody.host like 'conference.%'
    and prosody.key like '%@%'
    and prosody.value in ('owner','admin','member','outcast','none');

-- room_nicknames is obsolete (read key='_occupants', which no longer exists).
drop view if exists room_nicknames cascade;

-- =============================================================================
-- RULES — on `prosodyarchive`
-- =============================================================================

-- mentions
create or replace rule update_muc_mention as
  on insert to prosodyarchive where NEW.store = 'muc_log'
  do insert into totalmentions (username, target, key, type) select
    split_part(split_part(jsonb_array_elements(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']])))::text, 'xmpp:',2), '"',1) as username,
    NEW.user || '@' || NEW.host as target,
    NEW.key as key,
    'groupchat' as type;

create or replace rule update_muc_remote_mention as
  on insert to prosodyarchive where NEW.store = 'muc_remote'
  do insert into totalmentions (username, target, key, type) select
    split_part(split_part(jsonb_array_elements(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']])))::text, 'xmpp:',2), '"',1) as username,
    NEW.with as target,
    NEW.key as key,
    'groupchat' as type;

-- NOTE: the legacy `update_room_nick_jid_map` rule (which read the now-dropped
-- `room_nicknames` view) is GONE. room_nick_jid_map is maintained by the
-- `fupdate_room_nick_jid` trigger on the `config` store plus the invite rules
-- below.

-- room_nick_jid_map from invite messages (unchanged — XML-parsed)
create or replace rule update_room_nick_jid_map_invitee as
  on insert to prosodyarchive where NEW.store = 'archive' and (NEW.with like '%@conference.%')
  do 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,
    (case
      when ((xpath('/message/x:x/x:invite/x:reason/text() | /jc:message/x:x/x:invite/x:reason/text()', NEW.value::xml, ARRAY[ARRAY['x', 'http://jabber.org/protocol/muc#user'], ARRAY['jc', 'jabber:client']]))[1]::text  = '1') then 0
      else extract(epoch from now())::integer
    end) as since;

create or replace rule update_room_nick_jid_map_inviter as
  on insert to prosodyarchive where NEW.store = 'archive' and (NEW.with like '%@conference.%')
  do 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(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((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',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;

create or replace rule update_room_nick_jid_map_remote_inv as
  on insert to prosodyarchive where NEW.store = 'muc_remote_inv'
  do insert into room_nick_jid_map (room_name, user_jid, nickname, since)
  select
    NEW.user||'@'||NEW.host as room_name,
    NEW.with as user_jid,
    NEW.user||'@'||NEW.host||'/'||NEW.with as nickname,
    extract(epoch from now())::integer;

create or replace rule init_read_conversation_invitee as
  on insert to prosodyarchive where NEW.store = 'archive' and (NEW.with like '%@conference.%')
  do insert into read_conversation (username, target, timestamp)
  select
    (split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1)) as username,
    (split_part((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1)) as target,
    (extract(epoch from now())::integer - 60) as timestamp;

-- single-chat recent history
create or replace rule update_single_chat_recent as
  on insert to prosodyarchive where NEW.store = 'archive'
  do 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,
    NEW.when as timestamp,
    (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1] as message,
    (xpath('/message/@id | /jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS message_id,
    (xpath('//originalMessage | //xc:originalMessage', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as original_message,
    (xpath('//attachment | //xc:attachment', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_attachment,
    (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_conference,
    (case
      when (split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) = NEW.user || '@' || NEW.host) then true
      else false
      end) as incoming,
    NULL as received_receipt,
    NEW.when as updated_at,
    true as has_data,
    false as deleted
  where
    NEW.store='archive'
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
    AND (xpath('//message/nsx:x/nsx:invite/@from'::text, new.value::xml, ARRAY[ARRAY['nsx'::text, 'http://jabber.org/protocol/muc#user'::text]]))[1]::text IS NULL
    and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NULL
    and ( (
      (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] IS NOT NULL AND
      (xpath('//vncTalkConference/conferenceId/text() | //xc:vncTalkConference/xc:conferenceId/text()', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1]::text ilike '%#%'
      ) or (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] IS NULL )
    AND NOT (NEW.with like '%@conference.%')
    AND NOT (NEW.with = NEW.user || '@' || NEW.host)
    order by target,timestamp desc;

-- broadcast recent history
create or replace rule update_broadcast_recent as
  on insert to prosodyarchive where NEW.store = 'archive'
  do 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,
    (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as target,
    'broadcast' as type,
    NEW.when as timestamp,
    (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1] as message,
    (xpath('/message/@id | /jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS message_id,
    (xpath('//originalMessage | //xc:originalMessage', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as original_message,
    (xpath('//attachment | //xc:attachment', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_attachment,
    (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_conference,
    (xpath('//message/xc:vncTalkBroadcast/@title', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1] as broadcast_title,
    (case
      when (split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) = NEW.user || '@' || NEW.host) then true
      else false
      end) as incoming,
    NULL as received_receipt,
    NEW.when as updated_at,
    true as has_data,
    false as deleted
  where
    NEW.store='archive'
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
    AND (xpath('//message/nsx:x/nsx:invite/@from'::text, new.value::xml, ARRAY[ARRAY['nsx'::text, 'http://jabber.org/protocol/muc#user'::text]]))[1]::text IS NULL
    and (xpath('//message/xc:vncTalkBroadcast'::text, new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']])) IS NOT NULL
    and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NOT NULL
    AND NOT (NEW.with like '%@conference.%')
    order by timestamp desc;

-- broadcast audience
create or replace rule update_broadcast_audience as
  on insert to prosodyarchive where NEW.store = 'archive'
  do insert into broadcast_audience (broadcast_owner, broadcast_target, audience, title, tags, description) select
    NEW.with as broadcast_owner,
    (xpath('//message/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1] as broadcast_target,
    array_to_json(xpath('//message/xc:vncTalkBroadcast/xc:to/text()'::text, NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))::jsonb as audience,
    (xpath('//message/xc:vncTalkBroadcast/@title', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as title,
    array_to_json(xpath('//message/xc:vncTalkBroadcast/xc:tag/text()'::text, NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))::jsonb as tags,
    (xpath('//message/xc:vncTalkBroadcast/@description', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as description
  where
    NEW.store='archive'
    AND (xpath('//message/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1] IS NOT NULL
    and (xpath('//message/xc:vncTalkBroadcast'::text, new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']])) IS NOT NULL
    ;

create or replace rule update_broadcast_audience_remote as
  on insert to prosodyarchive where NEW.store = 'archive'
  do insert into broadcast_audience (broadcast_owner, broadcast_target, audience, title) select
    NEW.with as broadcast_owner,
    (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1] as broadcast_target,
    '[]' as audience,
    (xpath('//message/xc:vncTalkBroadcast/@title', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as title
  WHERE
    NEW.store='archive'
    AND (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1] IS NOT NULL
    and (xpath('//message/xc:vncTalkBroadcast'::text, NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']])) IS NOT NULL
    and (split_part(NEW.with, '@', 2) <> NEW.host);

-- groupchat recent history (uses room_membership view — see VIEWS section)
create or replace rule update_muc_recent as
  on insert to prosodyarchive where NEW.store = 'muc_log'
  do 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,
    NEW.when as timestamp,
    (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1] as message,
    (xpath('/message/@id | /jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS message_id,
    (xpath('//originalMessage | //xc:originalMessage', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as original_message,
    (xpath('//attachment | //xc:attachment', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_attachment,
    (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_conference,
    (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']])))::text as mentions,
    (case
      when ((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text = room_membership.user_room_nickname) then false
      else true
     end) as incoming,
    false as received_receipt,
    NEW.when 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
    from room_membership where NEW.host=room_membership.host and NEW.user=room_membership.room and NEW.store='muc_log'
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
    order by username, room_membership.user_room_nickname desc;

create or replace rule update_muc_remote_recent as
  on insert to prosodyarchive where NEW.store = 'muc_remote'
  do insert into recent_history_table (username, target, type, timestamp, message, message_id, original_message, x_attachment, x_conference, mentions, incoming, received_receipt, updated_at, has_data) select
    NEW.user || '@' || NEW.host as username,
    NEW.with as target,
    'groupchat' as type,
    NEW.when as timestamp,
    (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1] as message,
    (xpath('/message/@id | /jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS message_id,
    (xpath('//originalMessage | //xc:originalMessage', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as original_message,
    (xpath('//attachment | //xc:attachment', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_attachment,
    (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1] as x_conference,
    (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']])))::text as mentions,
    (case
      when ((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text = room_nick_jid_map.nickname) then false
      else true
     end) as incoming,
    false as received_receipt,
    NEW.when as updated_at,
    true as has_data
    from room_nick_jid_map where NEW.with=room_nick_jid_map.room_name and (room_nick_jid_map.user_jid = (NEW.user || '@' || NEW.host)) and NEW.store='muc_remote'
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
    order by username, room_nick_jid_map.nickname desc;

drop rule if exists update_receipt_recent on prosodyarchive cascade;

-- mark conversations read
create or replace rule mark_single_chat_read as
  on insert to prosodyarchive where NEW.store = 'archive'
  AND split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) = NEW.with
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
  do insert into read_conversation (username, target, "timestamp")
    SELECT
      (new."user" || '@'::text) || new.host AS username,
      NEW.with as target,
      NEW.when as timestamp
    where split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) = NEW.with;

create or replace rule mark_muc_read as
  on insert to prosodyarchive where NEW.store = 'muc_log'
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
  AND (xpath('//vncTalkConference | //xc:vncTalkConference', NEW.value::xml, ARRAY[ARRAY['xc','xmpp:vnctalk']]))[1]::text IS NULL
  do insert into read_conversation (username, target, timestamp)  select
    room_nick_jid_map.user_jid AS username,
    (new."user" || '@'::text) || new.host AS target,
     new."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 = (xpath('//message/@from | //jc:message/@from'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text
  ;

-- unread message ids
create or replace rule insert_unread_message_ids_chat as
  on insert to prosodyarchive where NEW.store = 'archive'
  do insert into unread_message_ids (message_id, sender, receipient, timestamp)
    select
    (case
      when (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text IS NULL then NEW.key
      else  (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text
     end) as message_id,
      NEW.with as sender,
      split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) as receipient,
      NEW.when as timestamp
  where split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) <> NEW.with
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
  AND (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 IS NULL
  AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
  and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NULL
  AND NOT (NEW.with like '%@conference.%')
  ;

create or replace rule insert_unread_message_ids_broadcast as
  on insert to prosodyarchive where NEW.store = 'archive'
  do insert into unread_message_ids (message_id, sender, receipient, timestamp)
    select
    (case
      when (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text IS NULL then NEW.key
      else  (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text
     end) as message_id,
      (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as receipient,
      split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) as receipient,
      NEW.when as timestamp
  where split_part((xpath('//message/@to | //jabberclient:message/@to', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',1) <> NEW.with
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
  AND (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 IS NULL
  AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
  and (xpath('//message/xc:vncTalkBroadcast'::text, new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']])) IS NOT NULL
  and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NOT NULL
  AND NOT (NEW.with like '%@conference.%')
  ;

create or replace rule remove_unread_message_ids_deleted_chat as
    on insert to prosodyarchive where NEW.store = 'archive' and (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 IS NOT NULL
    do delete from unread_message_ids where message_id=(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;

create or replace rule insert_unread_message_ids_muc as
  on insert to prosodyarchive where NEW.store = 'muc_log'
  do insert into unread_message_ids (receipient, message_id, sender, timestamp)
    select
    DISTINCT room_membership.username AS receipient,
    (case
      when (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text IS NULL then NEW.key
      else  (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text
     end) as message_id,
      NEW.user || '@' || NEW.host as sender,
      NEW.when::integer as timestamp
    from room_membership
    where room_membership.host=NEW.host and room_membership.room=NEW.user and room_membership.username not
      in (select username from room_membership where user_room_nickname=((xpath('//message/@from | //jc:message/@from'::text, NEW.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text))
      AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
      AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
      AND (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 IS NULL
      AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
;

create or replace rule remove_unread_message_ids_deleted_muc as
  on insert to prosodyarchive where NEW.store = 'muc_log' and (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 IS NOT NULL
  do delete from unread_message_ids where message_id=(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;

create or replace rule insert_unread_message_ids_remotemuc as
  on insert to prosodyarchive where NEW.store = 'muc_remote'
  do insert into unread_message_ids (receipient, message_id, sender, timestamp)
  select
  NEW.user || '@' || NEW.host as receipient,
  NEW.key as message_id,
  NEW.with as sender,
  NEW.when as timestamp
where (split_part((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',2) <> NEW.user || '@' || NEW.host)
AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
AND (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 IS NULL
AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NULL
AND (NEW.with like '%@conference.%')
;

create or replace rule remove_unread_message_ids_deleted_remotemuc as
  on insert to prosodyarchive where NEW.store = 'muc_remote' and (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 IS NOT NULL
  do delete from unread_message_ids where message_id=(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;

-- unread mention ids
create or replace rule insert_unread_mention_ids_muc as
  on insert to prosodyarchive where NEW.store = 'muc_log'
  do insert into unread_message_mention_ids (receipient, message_id, sender, timestamp)
    select
    DISTINCT room_membership.username AS receipient,
    (case
      when (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text IS NULL then NEW.key
      else  (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text
     end) as message_id,
      NEW.user || '@' || NEW.host as sender,
      NEW.when::integer as timestamp
    from room_membership
    where room_membership.host=NEW.host and room_membership.room=NEW.user and room_membership.username not
      in (select username from room_membership where user_room_nickname=((xpath('//message/@from | //jc:message/@from'::text, NEW.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text))
      AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
      AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
      AND (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 IS NULL
      AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
      AND (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']])))::text like '%xmpp:'||room_membership.username||'%'
;

create or replace rule insert_unread_message_ids_remotemuc_mention as
  on insert to prosodyarchive where NEW.store = 'muc_remote'
  do insert into unread_message_mention_ids (receipient, message_id, sender, timestamp)
  select
  NEW.user || '@' || NEW.host as receipient,
  (case
    when (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text IS NULL then NEW.key
    else  (xpath('//message/@id | //jc:message/@id', NEW.value::xml, ARRAY[ARRAY['jc','jabber:client']]))[1]::text
  end) as message_id,
  NEW.with as sender,
  NEW.when as timestamp
where (split_part((xpath('//message/@from | //jabberclient:message/@from', NEW.value::xml, ARRAY[ARRAY['jabberclient','jabber:client']]))[1]::text, '/',2) <> NEW.user || '@' || NEW.host)
AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, new.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
AND (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 IS NULL
AND (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', new.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text IS NULL
and (xpath('//message/xc:vncTalkBroadcast/@origtarget', NEW.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text IS NULL
AND (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']])))::text like '%xmpp:'||NEW.user||'@'||NEW.host||'%'
AND (NEW.with like '%@conference.%')
;

-- recent_history upsert rules
create or replace rule instert_into_chat_recent as
  on insert to recent_history_table where ((exists (select 1 from recent_history_table where target=NEW.target and username=NEW.username)) and NEW.timestamp is not null)
  do instead
    update recent_history_table set
      type=NEW.type,
      message=NEW.message,
      original_message=NEW.original_message,
      x_attachment=NEW.x_attachment,
      x_conference=NEW.x_conference,
      timestamp=NEW.timestamp,
      incoming=NEW.incoming,
      message_id=NEW.message_id,
      received_receipt=null,
      updated_at=NEW.timestamp,
      has_data=true,
      broadcast_title=NEW.broadcast_title,
      sort_id=NEW.sort_id,
      deleted=false,
      mentions=NEW.mentions
    where target=NEW.target and username=NEW.username and ((recent_history_table.deleted = false) or (NEW.deleted IS NOT NULL));

create or replace rule instert_into_chat_recent_mute as
  on insert to recent_history_table where ((exists (select 1 from recent_history_table where target=NEW.target and username=NEW.username)) and (NEW.timestamp is null) and (NEW.mute_notification is null))
  do instead
    update recent_history_table set
      mute_sound = NEW.mute_sound,
      updated_at=EXTRACT(EPOCH FROM now())::integer
    where target=NEW.target and username=NEW.username;

create or replace rule instert_into_chat_recent_notify as
  on insert to recent_history_table where ((exists (select 1 from recent_history_table where target=NEW.target and username=NEW.username)) and (NEW.timestamp is null) and (NEW.mute_sound is null))
  do instead
    update recent_history_table set
      mute_notification = NEW.mute_notification,
      updated_at=EXTRACT(EPOCH FROM now())::integer
    where target=NEW.target and username=NEW.username;

-- =============================================================================
-- VIEWS
-- =============================================================================

create or replace view total_mentions as
select
  username,
  target,
  'groupchat' as type,
  count(*) as total
from totalmentions group by username, target;

create or replace view unread_conversation_ids as
 select
   read_conversation.username as username,
   read_conversation.target as target,
   jsonb_agg(unread_message_idjoin.message_id) as unreadids
 from read_conversation
 join unread_message_ids as unread_message_idjoin on
   ( unread_message_idjoin.receipient=read_conversation.username and unread_message_idjoin.sender=read_conversation.target and unread_message_idjoin.timestamp>read_conversation.timestamp)
  group by read_conversation.username, read_conversation.target
union
  select
  unread_message_ids.receipient as username,
  unread_message_ids.sender as target,
  jsonb_agg(unread_message_ids.message_id) as unreadids
from unread_message_ids
where unread_message_ids.receipient||'#'||unread_message_ids.sender not in (select username||'#'||target from read_conversation)
group by unread_message_ids.receipient, unread_message_ids.sender
;

create or replace view unread_conversation_counts as
 select
   read_conversation.username as username,
   read_conversation.target as target,
   count(unread_message_idjoin.message_id) as unreadidcount
 from read_conversation
 join unread_message_ids as unread_message_idjoin on
   ( unread_message_idjoin.receipient=read_conversation.username and unread_message_idjoin.sender=read_conversation.target and unread_message_idjoin.timestamp>read_conversation.timestamp)
  group by read_conversation.username, read_conversation.target
union
  select
  unread_message_ids.receipient as username,
  unread_message_ids.sender as target,
  count(unread_message_ids.message_id) as unreadidcount
from unread_message_ids
where unread_message_ids.receipient||'#'||unread_message_ids.sender not in (select username||'#'||target from read_conversation)
group by unread_message_ids.receipient, unread_message_ids.sender
;

create or replace view unread_mention_ids as
 select
   read_conversation.username as username,
   read_conversation.target as target,
   jsonb_agg(unread_message_mention_idjoin.message_id) as unreadids
 from read_conversation
 join unread_message_mention_ids as unread_message_mention_idjoin on
   ( unread_message_mention_idjoin.receipient=read_conversation.username and unread_message_mention_idjoin.sender=read_conversation.target and unread_message_mention_idjoin.timestamp>read_conversation.timestamp)
  group by read_conversation.username, read_conversation.target
union
  select
  unread_message_mention_ids.receipient as username,
  unread_message_mention_ids.sender as target,
  jsonb_agg(unread_message_mention_ids.message_id) as unreadids
from unread_message_mention_ids
where unread_message_mention_ids.receipient||'#'||unread_message_mention_ids.sender not in (select username||'#'||target from read_conversation)
group by unread_message_mention_ids.receipient, unread_message_mention_ids.sender
;

create or replace view user_notify_opt0 as
  select prosody.user||'@'||prosody.host as email from prosody where
  ( prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\\"0%'
  or prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\"0%' );

create or replace view user_notify_opt1 as
  select prosody.user||'@'||prosody.host as email from prosody where (
    prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\\"1%'
    or prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"__array":["\\"1%'
    or prosody.value like '%emailNotification"},"name":"document","__array":["\\"1%'
    or prosody.value like '%emailNotification"},"__array":["\\"1%'
    );

create or replace view user_notify_opt2 as
  select prosody.user||'@'||prosody.host as email from prosody where (
    prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\\"2%'
    or prosody.value like '%emailNotification"},"name":"document","__array":["\\"2%'
    or prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"__array":["\\"2%'
    or prosody.value like '%emailNotification"},"__array":["\\"2%');

create or replace view user_notify_opt3 as
  select prosody.user||'@'||prosody.host as email from prosody where (
    prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\\"3%'
    or prosody.value like '%emailNotification"},"name":"document","__array":["\\"3%'
    or prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"__array":["\\"3%'
    or prosody.value like '%emailNotification"},"__array":["\\"3%');

create or replace view user_notify_opt4 as
  select prosody.user||'@'||prosody.host as email from prosody where (
    prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"name":"document","__array":["\\"4%'
    or prosody.value like '%emailNotification"},"name":"document","__array":["\\"4%'
    or prosody.value like '%emailNotification","xmlns":"stanza:io:json"},"__array":["\\"4%'
    or prosody.value like '%emailNotification"},"__array":["\\"4%');

create or replace view digest_messages as
select
  prosodyarchive.user||'@'||prosodyarchive.host as email,
  prosodyarchive.when as when,
  (xpath('/message/@from | /jc:message/@from', prosodyarchive.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS from,
  (xpath('/message/body/text() | /message/jc:body/text() | /jc:message/jc:body/text()', prosodyarchive.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS body,
  (xpath('//message/@origtarget | /message/xc:vncTalkBroadcast/@origtarget', prosodyarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as broadcast,
  (xpath('/message/xc:attachment | /jc:message/xc:attachment', prosodyarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text AS x_attachment,
  (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', prosodyarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text as x_vncConference,
  (xpath('/message/@from | /jc:message/@from', prosodyarchive.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS target,
  NULL as enddtime,
  NULL as starttime,
  'chat' as type
from prosodyarchive
where prosodyarchive.store='offline' and  (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, prosodyarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
    AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, prosodyarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
union
select
  recent_history_table.username as email,
  mucarchive.when as when,
  (xpath('/message/@from | /jc:message/@from', mucarchive.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS from,
  (xpath('/message/body/text() | /message/jc:body/text() | /jc:message/jc:body/text()', mucarchive.value::xml, ARRAY[ARRAY['jc', 'jabber:client']]))[1]::text AS body,
  (xpath('//message/@origtarget | /message/xc:vncTalkBroadcast/@origtarget', mucarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk']]))[1]::text as broadcast,
  (xpath('/message/xc:attachment | /jc:message/xc:attachment', mucarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text AS x_attachment,
  (xpath('/message/xc:vncTalkConference | /jc:message/xc:vncTalkConference', mucarchive.value::xml, ARRAY[ARRAY['xc', 'xmpp:vnctalk'], ARRAY['jc', 'jabber:client']]))[1]::text as x_vncConference,
  recent_history_table.target as target,
  recent_history_table.timestamp as endtime,
  read_conversation.timestamp as starttime,
  'groupchat' as type
from
  recent_history_table, read_conversation
join prosodyarchive as mucarchive on (
      mucarchive.user = split_part(read_conversation.target, '@', 1) and
      mucarchive.host = split_part(read_conversation.target, '@', 2) and
      mucarchive.store='muc_log' and
      (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, mucarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text and
      mucarchive.when > read_conversation.timestamp and
      ((xpath('//message/@from | //jc:message/@from'::text, mucarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text
        not in (select nickname from room_nick_jid_map where room_name=read_conversation.target and user_jid=read_conversation.username))
    )
where
  recent_history_table.type='groupchat'
  AND recent_history_table.username=read_conversation.username
  AND recent_history_table.target=read_conversation.target
  AND (recent_history_table.timestamp > read_conversation.timestamp)
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, mucarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text <> ''::text
  AND (xpath('//body/text() | /jc:message/jc:body/text() | //jc:body/text()'::text, mucarchive.value::xml, ARRAY[ARRAY['jc'::text, 'jabber:client'::text]]))[1]::text IS NOT NULL
;

create or replace view room_membernames as SELECT room_nick_jid_map.room_name,
    jsonb_agg(room_nick_jid_map.user_jid) AS members,
    split_part(split_part(mucjoin.value, '"subject":'::text, 2), '"'::text, 2) AS title,
    jsonb_agg(
        CASE
            WHEN ((((((split_part(split_part(split_part(namejoin.value, 'NICKNAME","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'FN","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'GIVEN","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'FAMILY","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)) = '   '::text THEN namejoin."user"
            ELSE (((((split_part(split_part(split_part(namejoin.value, 'NICKNAME","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'FN","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'GIVEN","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)) || ' '::text) || split_part(split_part(split_part(namejoin.value, 'FAMILY","'::text, 2), '__array":["'::text, 2), '"]}'::text, 1)
        END) AS membernames
   FROM room_nick_jid_map
     LEFT JOIN prosody mucjoin ON room_nick_jid_map.room_name = ((mucjoin."user" || '@'::text) || mucjoin.host) AND mucjoin.store = 'config'::text AND mucjoin.key = '_data'::text AND mucjoin.value ~~ '%"subject":%'::text
     LEFT JOIN prosody namejoin ON room_nick_jid_map.user_jid = ((namejoin."user" || '@'::text) || namejoin.host) AND namejoin.store = 'vcard'::text AND namejoin.key = ''::text AND namejoin.type = 'json'::text
  GROUP BY room_nick_jid_map.room_name, mucjoin.value;

create or replace view inverseconfmap as
  select
    split_part(split_part(value, 'value":"',2), '","', 1) as meeting,
    conferencekey as key
  from conferencemap;

-- =============================================================================
-- TEXT SEARCH CONFIGURATION
-- (guarded so the script is idempotent even if the dictionary/template is missing)
-- =============================================================================

do $$
begin
  if not exists (select 1 from pg_ts_config where cfgname = 'english_nostop') then
    create text search configuration public.english_nostop ( parser = pg_catalog."default" );
  end if;
end $$;

-- Reconcile mappings idempotently: remove all existing mappings for this
-- configuration, then re-add the desired set (only if the dictionary exists).
-- NOTE: do NOT delete from pg_ts_config_map directly — it is a PostgreSQL
-- system catalog and only writable by superusers. ALTER ... DROP MAPPING
-- IF EXISTS is the non-superuser DDL that achieves the same effect.
do $$
begin
  alter text search configuration public.english_nostop drop mapping if exists
    for asciiword, word, hword_part, hword_asciipart, asciihword, hword,
         numword, email, url, host, sfloat, version, hword_numpart, numhword,
         url_path, file, "float", "int", uint;
  if exists (select 1 from pg_ts_dict where dictname = 'english_stem_nostop') then
    alter text search configuration public.english_nostop add mapping for asciiword with public.english_stem_nostop;
    alter text search configuration public.english_nostop add mapping for word with public.english_stem_nostop;
    alter text search configuration public.english_nostop add mapping for hword_part with public.english_stem_nostop;
    alter text search configuration public.english_nostop add mapping for hword_asciipart with public.english_stem_nostop;
    alter text search configuration public.english_nostop add mapping for asciihword with public.english_stem_nostop;
    alter text search configuration public.english_nostop add mapping for hword with public.english_stem_nostop;
  end if;
  alter text search configuration public.english_nostop add mapping for numword with simple;
  alter text search configuration public.english_nostop add mapping for email with simple;
  alter text search configuration public.english_nostop add mapping for url with simple;
  alter text search configuration public.english_nostop add mapping for host with simple;
  alter text search configuration public.english_nostop add mapping for sfloat with simple;
  alter text search configuration public.english_nostop add mapping for version with simple;
  alter text search configuration public.english_nostop add mapping for hword_numpart with simple;
  alter text search configuration public.english_nostop add mapping for numhword with simple;
  alter text search configuration public.english_nostop add mapping for url_path with simple;
  alter text search configuration public.english_nostop add mapping for file with simple;
  alter text search configuration public.english_nostop add mapping for "float" with simple;
  alter text search configuration public.english_nostop add mapping for "int" with simple;
  alter text search configuration public.english_nostop add mapping for uint with simple;
end $$;

-- =============================================================================
-- TRIGGER FUNCTIONS
-- =============================================================================

-- fprocess_messages — main message-processing trigger on prosodyarchive.
-- Body unchanged from 0.11.6; it relies on room_nick_jid_map (table) and
-- room_membership (view), both of which are kept correct by the 13.0.6
-- triggers/views above.
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;

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

-- fupdate_room_nick_jid — 13.0.6: fires on the `_data` row (always written on
-- every room save) and rebuilds room_nick_jid_map from the per-affiliation
-- rows. Nickname = "<room@host>/<bare-jid>" (bare JID is the only nickname
-- used by allowed clients).
create or replace function fupdate_room_nick_jid()
  returns trigger AS
  $BODY$
  DECLARE
    _room text;
  BEGIN
    _room := NEW.user || '@' || NEW.host;

    delete from room_nick_jid_map
    where room_name = _room
      and user_jid not in (
        select key from prosody
        where store = 'config'
          and host = NEW.host
          and "user" = NEW.user
          and key like '%@%'
          and value in ('owner','admin','member','outcast','none')
      );

    update recent_history_table
    set deleted = true
    where target = _room
      and username not in (
        select key from prosody
        where store = 'config'
          and host = NEW.host
          and "user" = NEW.user
          and key like '%@%'
          and value in ('owner','admin','member','outcast','none')
      );

    insert into room_nick_jid_map (room_name, user_jid, nickname, since)
    select
      _room,
      key,
      _room || '/' || key,
      (extract(epoch from now())::integer - 60)
    from prosody
    where store = 'config'
      and host = NEW.host
      and "user" = NEW.user
      and key like '%@%'
      and value in ('owner','admin','member','outcast','none')
      and not exists (
        select 1 from room_nick_jid_map r
        where r.room_name = _room
          and r.user_jid = prosody.key
          and r.nickname = _room || '/' || prosody.key
      );

    update room_nick_jid_map r
    set since = (extract(epoch from now())::integer - 60)
    from prosody
    where prosody.store = 'config'
      and prosody.host = NEW.host
      and prosody."user" = NEW.user
      and prosody.key like '%@%'
      and prosody.value in ('owner','admin','member','outcast','none')
      and r.room_name = _room
      and r.user_jid = prosody.key
      and r.nickname = _room || '/' || prosody.key;

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

drop trigger if exists update_room_nick_jid on prosody;
create trigger update_room_nick_jid after insert on prosody
  for each row
  when (NEW.store = 'config' and NEW.key = '_data' and NEW.host like 'conference.%')
  execute procedure fupdate_room_nick_jid();

-- fupdate_group_owners — 13.0.6: replaces the legacy `update_group_owners`
-- rule (which keyed off `_affiliations` and conflicted with the
-- `upsert_group_owners` INSTEAD rule). Fires on the `_data` row and
-- reconciles group_owners from the per-affiliation owner rows.
create or replace function fupdate_group_owners()
  returns trigger AS
  $BODY$
  DECLARE
    _room text;
  BEGIN
    _room := NEW.user || '@' || NEW.host;

    delete from group_owners
    where room = _room
      and owner not in (
        select key from prosody
        where store = 'config'
          and host = NEW.host
          and "user" = NEW.user
          and key like '%@%'
          and value = 'owner'
      );

    insert into group_owners (room, owner, created, updated)
    select
      _room,
      sub.owner,
      extract(epoch from now())::integer,
      extract(epoch from now())::integer
    from (
      select key as owner
      from prosody
      where store = 'config'
        and host = NEW.host
        and "user" = NEW.user
        and key like '%@%'
        and value = 'owner'
      order by key
      limit 1
    ) sub
    where not exists (
      select 1 from group_owners go where go.room = _room
    );

    update group_owners
    set owner = sub.owner,
        updated = extract(epoch from now())::integer
    from (
      select key as owner
      from prosody
      where store = 'config'
        and host = NEW.host
        and "user" = NEW.user
        and key like '%@%'
        and value = 'owner'
      order by key
      limit 1
    ) sub
    where group_owners.room = _room
      and group_owners.owner <> sub.owner;

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

drop trigger if exists update_group_owners on prosody;
create trigger update_group_owners after insert on prosody
  for each row
  when (NEW.store = 'config' and NEW.key = '_data' and NEW.host like 'conference.%')
  execute procedure fupdate_group_owners();

-- fupdate_room_nick_jid_remote (unchanged — VNCtalk muc_remote store)
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;

drop trigger if exists update_room_nick_jid_remote on prosody;
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();

-- fupdate_mention_stamp (unchanged)
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;

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

-- fupdate_recent_avatar (unchanged — avatarupdate store)
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;

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

-- fupdate_iom_callstates (unchanged — callactive store)
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;

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

-- fupdate_msg_reactions / fupdate_msg_delreactions (unchanged — emojireactions)
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;

drop trigger if exists update_msg_reactions on emojireactions;
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;

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

-- fupdate_groupchat (unchanged — _data row still written by 13.0.6)
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;

drop trigger if exists update_groupchat on prosody;
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();