mattermost-mobile-upstream/app/actions/websocket/posts.ts

// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.

import {DeviceEventEmitter} from 'react-native';

import {storeMyChannelsForTeam, markChannelAsUnread, markChannelAsViewed, updateLastPostAt} from '@actions/local/channel';
import {addPostAcknowledgement, markPostAsDeleted, removePostAcknowledgement, updatePostTranslation} from '@actions/local/post';
import {captureRedactionEpoch} from '@actions/local/redaction';
import {createThreadFromNewPost, updateThread} from '@actions/local/thread';
import {getCurrentUserLocale} from '@actions/local/user';
import {fetchChannelStats, fetchMyChannel} from '@actions/remote/channel';
import {fetchPostAuthors, fetchPostById} from '@actions/remote/post';
import {openChannelIfNeeded} from '@actions/remote/preference';
import {fetchThread} from '@actions/remote/thread';
import {fetchMissingProfilesByIds} from '@actions/remote/user';
import {settleStreamedPost} from '@agents/actions/websocket';
import streamingStore from '@agents/store/streaming_store';
import {isAgentPost} from '@agents/utils';
import {ActionType, Events, Screens} from '@constants';
import {PostTypes} from '@constants/post';
import DatabaseManager from '@database/manager';
import {getChannelById, getMyChannel} from '@queries/servers/channel';
import {getPostById, syncPermalinkPreviewsForEditedPost} from '@queries/servers/post';
import {getCurrentChannelId, getCurrentTeamId, getCurrentUserId} from '@queries/servers/system';
import {getIsCRTEnabled} from '@queries/servers/thread';
import EphemeralStore from '@store/ephemeral_store';
import {NavigationStore} from '@store/navigation_store';
import {hasArrayChanged, isTablet} from '@utils/helpers';
import {logDebug, logWarning} from '@utils/log';
import {isFromWebhook, isPostEphemeral, isSystemMessage, restoreEphemeralIdentityFieldsForEdit, shouldIgnorePost} from '@utils/post';

import type {Model} from '@nozbe/watermelondb';
import type {PostWithRedactionEpoch} from '@typings/database/database';
import type MyChannelModel from '@typings/database/models/servers/my_channel';

function preparedMyChannelHack(myChannel: MyChannelModel) {
    if (!myChannel._preparedState) {
        myChannel._preparedState = null;
    }
}

const UNSANITIZED_BROADCAST_POST_TYPES = new Set<string>([PostTypes.BURN_ON_READ, PostTypes.EPHEMERAL, PostTypes.EPHEMERAL_ADD_TO_CHANNEL]);

/**
 * `posted` and `post_edited` payloads are redacted per recipient by the server's `abac_files`
 * broadcast hook, so they can be stamped as verified. Burn-on-read is outside that hook and at least
 * one ephemeral emitter bypasses it; returning undefined leaves those at their stored epoch.
 *
 * Call it as soon as the payload is parsed: it vouches for the decision the server made when it
 * broadcast, and an invalidation flushed while the handler awaits the network would otherwise be
 * stamped onto that older decision.
 */
const captureEpochForBroadcastPost = async (serverUrl: string, post: Post) => {
    if (UNSANITIZED_BROADCAST_POST_TYPES.has(post.type)) {
        return undefined;
    }
    return captureRedactionEpoch(serverUrl);
};

export async function handleNewPostEvent(serverUrl: string, msg: WebSocketMessage) {
    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
    if (!operator) {
        return;
    }

    const {database} = operator;

    let post: Post;
    try {
        post = JSON.parse(msg.data.post);
    } catch {
        return;
    }
    let redactionVerifiedEpoch = await captureEpochForBroadcastPost(serverUrl, post);
    const currentUserId = await getCurrentUserId(database);

    const existing = await getPostById(database, post.pending_post_id) || await getPostById(database, post.id);

    if (existing) {
        return;
    }

    const isCRTEnabled = await getIsCRTEnabled(database);
    if (isCRTEnabled) {
        await createThreadFromNewPost(serverUrl, post, false);
    }

    // Ensure the channel membership
    let myChannel = await getMyChannel(database, post.channel_id);
    if (myChannel) {
        const isCrtReply = isCRTEnabled && post.root_id !== '';

        // Don't change lastPostAt if the post is thread post
        if (!isCrtReply) {
            const {member} = await updateLastPostAt(serverUrl, post.channel_id, post.create_at, false);
            if (member) {
                myChannel = member;
            }
        }
    } else {
        const myChannelRequest = await fetchMyChannel(serverUrl, '', post.channel_id, true);
        if (myChannelRequest.error) {
            return;
        }

        // We want to have this on the database so we can make any needed update later
        const myChannelModels = await storeMyChannelsForTeam(serverUrl, '', myChannelRequest.channels!, myChannelRequest.memberships!, false);
        if (myChannelModels.error) {
            return;
        }

        myChannel = await getMyChannel(database, post.channel_id);
        if (!myChannel) {
            return;
        }
    }

    // If we don't have the root post for this post, fetch it from the server
    if (post.root_id) {
        const rootPost = await getPostById(database, post.root_id);

        if (!rootPost) {
            fetchPostById(serverUrl, post.root_id);
        }
    }

    const currentChannelId = await getCurrentChannelId(database);

    if (post.channel_id === currentChannelId) {
        const data = {
            channelId: post.channel_id,
            rootId: post.root_id,
            userId: post.user_id,
            now: Date.now(),
        };
        DeviceEventEmitter.emit(Events.USER_STOP_TYPING, data);
    }

    const models: Model[] = [];

    const {authors} = await fetchPostAuthors(serverUrl, [post], true);
    if (authors?.length) {
        const authorsModels = await operator.handleUsers({users: authors, prepareRecordsOnly: true});
        models.push(...authorsModels);
    }

    if (!shouldIgnorePost(post)) {
        let markAsViewed = false;

        if (!myChannel.manuallyUnread) {
            if (
                post.user_id === currentUserId &&
                !isSystemMessage(post) &&
                !isFromWebhook(post)
            ) {
                markAsViewed = true;
            } else if ((post.channel_id === currentChannelId)) {
                const isChannelScreenMounted = NavigationStore.getScreensInStack().includes(Screens.CHANNEL);

                const isTabletDevice = isTablet();
                if (isChannelScreenMounted || isTabletDevice) {
                    markAsViewed = false;
                }
            }
        }

        if (markAsViewed) {
            preparedMyChannelHack(myChannel);
            const {member: viewedAt} = await markChannelAsViewed(serverUrl, post.channel_id, false, true);
            if (viewedAt) {
                models.push(viewedAt);
            }
        } else if (!isCRTEnabled || !post.root_id) {
            const hasMentions = Boolean(msg.data.mentions?.includes(currentUserId));
            const hasUrgentMention = hasMentions && post.metadata?.priority?.priority === 'urgent';
            preparedMyChannelHack(myChannel);
            const {member: unreadAt} = await markChannelAsUnread(
                serverUrl,
                {
                    channelId: post.channel_id,
                    messageCount: myChannel.messageCount + 1,
                    mentionsCount: myChannel.mentionsCount + (hasMentions ? 1 : 0),
                    urgentMentionCount: myChannel.urgentMentionCount + (hasUrgentMention ? 1 : 0),
                    lastViewed: myChannel.lastViewedAt,
                },
                true,
            );
            if (unreadAt) {
                models.push(unreadAt);
            }
        }

        openChannelIfNeeded(serverUrl, post.channel_id);
    }

    let actionType: string = ActionType.POSTS.RECEIVED_NEW;
    if (isCRTEnabled && post.root_id) {
        actionType = ActionType.POSTS.RECEIVED_IN_THREAD;
    }

    const outOfOrderWebsocketEvent = EphemeralStore.getLastPostWebsocketEvent(serverUrl, post.id);
    if (outOfOrderWebsocketEvent?.deleted) {
        for (const model of models) {
            if (model._preparedState === 'update') {
                model.cancelPrepareUpdate();
            }
        }
        return;
    }

    if (outOfOrderWebsocketEvent?.post) {
        const editedPost = outOfOrderWebsocketEvent.post;
        if (post.type === PostTypes.EPHEMERAL || post.type === PostTypes.EPHEMERAL_ADD_TO_CHANNEL) {
            post = restoreEphemeralIdentityFieldsForEdit(editedPost, post);
        } else {
            post = editedPost;
        }

        // The stored payload is the edit, so the epoch is the one captured when the edit arrived.
        redactionVerifiedEpoch = (editedPost as PostWithRedactionEpoch).redaction_verified_epoch;
    }

    const postModels = await operator.handlePosts({
        actionType,
        order: [post.id],
        posts: [post],
        prepareRecordsOnly: true,
        redactionVerifiedEpoch,
    });

    models.push(...postModels);

    await operator.batchRecords(models, 'handleNewPostEvent');
}

export async function handlePostEdited(serverUrl: string, msg: WebSocketMessage) {
    const operator = DatabaseManager.serverDatabases[serverUrl]?.operator;
    if (!operator) {
        return;
    }
    const {database} = operator;

    let post: Post;
    try {
        post = JSON.parse(msg.data.post);
    } catch {
        return;
    }
    const redactionVerifiedEpoch = await captureEpochForBroadcastPost(serverUrl, post);

    const permalinkModels: Model[] = [];
    try {
        const updatedPermalinkPosts = await syncPermalinkPreviewsForEditedPost(database, post);
        if (updatedPermalinkPosts.length) {
            permalinkModels.push(...updatedPermalinkPosts);
        }
    } catch (error) {
        logWarning('Failed to sync permalink previews for edited post:', error);
    }

    const oldPost = await getPostById(database, post.id);
    if (!oldPost) {
        EphemeralStore.addEditingPost(serverUrl, {...post, redaction_verified_epoch: redactionVerifiedEpoch} as PostWithRedactionEpoch);

        // If we have permalink updates but no post to process, batch just the permalinks
        if (permalinkModels.length) {
            await operator.batchRecords(permalinkModels, 'handlePostEdited - permalink sync only');
        }
        return;
    }

    const models: Model[] = [...permalinkModels];

    if (isPostEphemeral(oldPost)) {
        post = restoreEphemeralIdentityFieldsForEdit(post, {
            create_at: oldPost.createAt,
            root_id: oldPost.rootId,
            user_id: oldPost.userId,
        });
    }

    const oldFileIds = oldPost.metadata?.files?.map((f) => f.id).filter((id): id is string => Boolean(id)) || [];
    const newFileIds = post.file_ids || [];
    const filesChanged = hasArrayChanged(oldFileIds, newFileIds);

    if (oldPost.isPinned !== post.is_pinned || filesChanged) {
        fetchChannelStats(serverUrl, post.channel_id);
    }

    const {authors} = await fetchPostAuthors(serverUrl, [post], true);
    if (authors?.length) {
        const authorsModels = await operator.handleUsers({users: authors, prepareRecordsOnly: true});
        models.push(...authorsModels);
    }

    let actionType: string = ActionType.POSTS.RECEIVED_NEW;
    const isCRTEnabled = await getIsCRTEnabled(database);
    if (isCRTEnabled && post.root_id) {
        actionType = ActionType.POSTS.RECEIVED_IN_THREAD;
    }

    const postModels = await operator.handlePosts({
        actionType,
        order: [post.id],
        posts: [post],
        prepareRecordsOnly: true,
        redactionVerifiedEpoch,
    });
    models.push(...postModels);

    await operator.batchRecords(models, 'handlePostEdited');

    // Conversation-backed agent posts are edited before the stream's `end`;
    // their streaming state is dropped once the post-stream refetch lands.
    // Websocket handlers aren't awaited, so `end` may already have been
    // handled before this post was stored; settle a finished stream again.
    if (isAgentPost(post)) {
        if (!post.props?.conversation_id) {
            streamingStore.removePost(serverUrl, post.id);
        } else if (streamingStore.getStreamingState(serverUrl, post.id) && !streamingStore.isStreaming(serverUrl, post.id)) {
            settleStreamedPost(serverUrl, post.id);
        }
    }
}

export async function handlePostDeleted(serverUrl: string, msg: WebSocketMessage) {
    try {
        const {database, operator} = DatabaseManager.getServerDatabaseAndOperator(serverUrl);

        const post: Post = JSON.parse(msg.data.post);

        const oldPost = await getPostById(database, post.id);
        if (!oldPost) {
            EphemeralStore.addRemovingPost(serverUrl, post.id);
            return;
        }

        const models: Model[] = [];

        const {model: deleteModel} = await markPostAsDeleted(serverUrl, post, true);
        if (deleteModel) {
            models.push(deleteModel);
        }

        // Check if the deleted post had files to refresh channel stats
        const hadFiles = (oldPost.metadata?.files?.length || 0) > 0;
        if (hadFiles) {
            fetchChannelStats(serverUrl, post.channel_id);
        }

        // update thread when a reply is deleted and CRT is enabled
        if (post.root_id) {
            const isCRTEnabled = await getIsCRTEnabled(database);
            if (isCRTEnabled) {
                // Update reply_count of the thread;
                // Note: reply_count includes current deleted count, So subtract 1 from reply_count
                const {model: threadModel} = await updateThread(serverUrl, post.root_id, {reply_count: post.reply_count - 1}, true);
                if (threadModel) {
                    models.push(threadModel);
                }

                const channel = await getChannelById(database, post.channel_id);
                if (channel) {
                    let {teamId} = channel;
                    if (!teamId) {
                        teamId = await getCurrentTeamId(database); // In case of DM/GM
                    }
                    fetchThread(serverUrl, teamId, post.root_id);
                }
            }
        }

        if (models.length) {
            await operator.batchRecords(models, 'handlePostDeleted');
        }
    } catch {
        // Do nothing
    }
}

export async function handlePostUnread(serverUrl: string, msg: WebSocketMessage) {
    const {channel_id: channelId, team_id: teamId} = msg.broadcast;
    const {
        mention_count: mentionCount,
        mention_count_root: mentionCountRoot,
        msg_count: msgCount,
        msg_count_root: msgCountRoot,
        last_viewed_at: lastViewedAt,
    } = msg.data;

    const database = DatabaseManager.serverDatabases[serverUrl]?.database;
    if (!database) {
        return;
    }
    const [myChannel, isCRTEnabled] = await Promise.all([
        getMyChannel(database, channelId),
        getIsCRTEnabled(database),
    ]);

    let messages = msgCount;
    let mentions = mentionCount;
    if (isCRTEnabled) {
        messages = msgCountRoot;
        mentions = mentionCountRoot;
    }

    if (!myChannel?.manuallyUnread) {
        const {channels, memberships, error} = await fetchMyChannel(serverUrl, teamId, channelId, true);
        const channel = channels?.[0];
        const membership = memberships?.[0];
        if (error || !membership) {
            logDebug('handlePostUnread', 'skipping unread update, fetchMyChannel failed or membership absent', channelId);
            return;
        }
        const postNumber = isCRTEnabled ? channel?.total_msg_count_root : channel?.total_msg_count;
        const delta = postNumber ? postNumber - messages : messages;

        markChannelAsUnread(serverUrl, {
            channelId,
            messageCount: delta,
            mentionsCount: mentions,
            urgentMentionCount: membership.urgent_mention_count ?? 0,
            lastViewed: lastViewedAt,
        });
    }
}

export async function handlePostAcknowledgementAdded(serverUrl: string, msg: WebSocketMessage) {
    try {
        const acknowledgement = JSON.parse(msg.data.acknowledgement);
        const {user_id, post_id, acknowledged_at} = acknowledgement;
        const database = DatabaseManager.serverDatabases[serverUrl]?.database;
        if (!database) {
            return;
        }
        const currentUserId = await getCurrentUserId(database);
        if (EphemeralStore.isAcknowledgingPost(post_id) && currentUserId === user_id) {
            return;
        }

        addPostAcknowledgement(serverUrl, post_id, user_id, acknowledged_at);
        fetchMissingProfilesByIds(serverUrl, [user_id]);
    } catch (error) {
        // Do nothing
    }
}

export async function handlePostAcknowledgementRemoved(serverUrl: string, msg: WebSocketMessage) {
    try {
        const acknowledgement = JSON.parse(msg.data.acknowledgement);
        const {user_id, post_id} = acknowledgement;
        const database = DatabaseManager.serverDatabases[serverUrl]?.database;
        if (!database) {
            return;
        }
        const currentUserId = await getCurrentUserId(database);
        if (EphemeralStore.isUnacknowledgingPost(post_id) && currentUserId === user_id) {
            return;
        }
        await removePostAcknowledgement(serverUrl, post_id, user_id);
    } catch (error) {
        // Do nothing
    }
}

export async function handlePostTranslationUpdatedEvent(serverUrl: string, msg: WebSocketMessage) {
    try {
        const translationData: PostTranslationUpdateData = msg.data;
        const locale = await getCurrentUserLocale(serverUrl);
        const myLanguageTranslation = translationData.translations[locale];

        if (!myLanguageTranslation) {
            return;
        }

        const translationObject: PostTranslation['object'] = myLanguageTranslation?.translation ? JSON.parse(myLanguageTranslation.translation) : undefined;
        await updatePostTranslation(serverUrl, translationData.object_id, locale, {
            object: translationObject,
            state: myLanguageTranslation.state,
            source_lang: myLanguageTranslation.src_lang,
        });
    } catch (error) {
        // Do nothing
    }
}