From 11c602a58b2a7b8f5824c9bda845075a59270d5e Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Fri, 11 Sep 2026 14:46:01 +0000 Subject: [PATCH] fixes --- .../src/Broadcast/Bot.hs | 1 - plans/2026-09-11-feed-jobs-split.md | 153 ++++++++++++++++++ src/Simplex/Chat/Delivery.hs | 28 ++-- src/Simplex/Chat/Library/Commands.hs | 67 +++----- src/Simplex/Chat/Library/Internal.hs | 9 -- src/Simplex/Chat/Library/Subscriber.hs | 90 +++++------ src/Simplex/Chat/Store/Delivery.hs | 59 +++---- src/Simplex/Chat/Store/Feeds.hs | 59 +++---- src/Simplex/Chat/Store/Messages.hs | 142 ++++------------ tests/Bots/BroadcastTests.hs | 1 - tests/ChatTests/Feed.hs | 8 - tests/ProtocolTests.hs | 6 + 12 files changed, 333 insertions(+), 290 deletions(-) create mode 100644 plans/2026-09-11-feed-jobs-split.md diff --git a/apps/simplex-broadcast-bot/src/Broadcast/Bot.hs b/apps/simplex-broadcast-bot/src/Broadcast/Bot.hs index cbc43ae37e..47b180206b 100644 --- a/apps/simplex-broadcast-bot/src/Broadcast/Bot.hs +++ b/apps/simplex-broadcast-bot/src/Broadcast/Bot.hs @@ -49,7 +49,6 @@ broadcastBot BroadcastBotOpts {publishers, welcomeMessage, prohibitedMessage} _u | allowContent mc -> void $ forkIO $ sendChatCmd cc (SendMessageBroadcast mc) >>= \case - -- delivery to the recipients continues in feed delivery jobs Right CRNewChatItems {} -> sendReply "Message is being delivered to all contacts" r -> putStrLn $ "Error broadcasting message: " <> show r | otherwise -> diff --git a/plans/2026-09-11-feed-jobs-split.md b/plans/2026-09-11-feed-jobs-split.md new file mode 100644 index 0000000000..230d504eb7 --- /dev/null +++ b/plans/2026-09-11-feed-jobs-split.md @@ -0,0 +1,153 @@ +# Split feed delivery from group delivery jobs + +Feed delivery is moved out of `delivery_jobs` into its own table, worker key and code path. +`delivery_jobs` returns to its pre-feed definition. +The feed migration is not released, so it is amended in place rather than superseded. + +## 1. Schema + +### delivery_jobs — revert + +Remove from `M20260905_feeds` (SQLite and Postgres): + +- `ALTER TABLE delivery_jobs ADD COLUMN feed_id` +- `ALTER TABLE delivery_jobs ADD COLUMN chat_item_id` +- `ALTER TABLE delivery_jobs ADD COLUMN message_ids` +- `ALTER TABLE delivery_jobs ADD COLUMN feed_cursor_id` +- `CREATE INDEX idx_delivery_jobs_feed_next` +- `CREATE INDEX idx_delivery_jobs_chat_item_id` +- the `PRAGMA writable_schema` rewrite of `group_id` (SQLite) +- `ALTER TABLE delivery_jobs ALTER COLUMN group_id DROP NOT NULL` (Postgres) +- the matching statements in both down migrations + +`delivery_jobs.group_id` stays `NOT NULL`. +`delivery_jobs.job_scope_spec_tag` holds `DeliveryJobSpecTag` only. +`delivery_jobs.worker_scope` holds `DeliveryWorkerScope` only. + +### feed_jobs — new table + +SQLite: + +``` +CREATE TABLE feed_jobs( + feed_job_id INTEGER PRIMARY KEY AUTOINCREMENT, + feed_id INTEGER NOT NULL REFERENCES feeds ON DELETE CASCADE, + chat_item_id INTEGER NOT NULL REFERENCES chat_items ON DELETE CASCADE, + worker_scope TEXT NOT NULL, + action_tag TEXT NOT NULL, + message_ids TEXT, + cursor_id INTEGER, + job_status TEXT NOT NULL, + job_err_reason TEXT, + failed INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT(datetime('now')), + updated_at TEXT NOT NULL DEFAULT(datetime('now')) +) STRICT; +CREATE INDEX idx_feed_jobs_next ON feed_jobs(feed_id, worker_scope, failed, job_status, feed_job_id); +CREATE INDEX idx_feed_jobs_chat_item_id ON feed_jobs(chat_item_id, action_tag); +``` + +Postgres uses `BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY`, `BIGINT`, `TIMESTAMPTZ NOT NULL DEFAULT (now())`, `SMALLINT NOT NULL DEFAULT 0` for `failed`. + +Down migration: `DROP INDEX idx_feed_jobs_chat_item_id`, `DROP INDEX idx_feed_jobs_next`, `DROP TABLE feed_jobs`. + +`worker_scope` holds `FeedWorkerScope`. +`action_tag` holds `FeedJobActionTag`. +`cursor_id` holds a contact id under `FWSContacts` and a group id under `FWSGroups`. + +## 2. Types + +`src/Simplex/Chat/Delivery.hs`: + +- delete `DeliveryJobKey` +- delete `DeliveryJobWork`, `DJWGroup`, `DJWFeed` +- restore `DeliveryJob` as the pre-feed record with `jobId`, `cursorId_`, `jobScope`, `senderGMIds`, `body` +- add `FeedJobKey = (FeedId, FeedWorkerScope)` +- add `FeedJob {feedJobId, feedItemId, feedAction, cursorId_}` +- keep `FeedWorkerScope`, `feedWorkerScopes`, `FeedJobAction`, `FeedJobActionTag`, `FeedInstanceSpec` and the action projections + +`src/Simplex/Chat/Controller.hs`: + +- `deliveryJobWorkers` reverts to `TMap DeliveryWorkerKey Worker` +- add `feedJobWorkers :: TMap FeedJobKey Worker` + +## 3. Store + +`src/Simplex/Chat/Store/Delivery.hs` reverts to its pre-feed content, with `getPendingDeliveryJobScopes` restored to the single `SELECT DISTINCT group_id, worker_scope` form. + +Feed job statements move to `src/Simplex/Chat/Store/Feeds.hs`: + +- `createFeedJobs` +- `getNextFeedJob` +- `updateFeedJobCursor` +- `completeFeedJob` +- `getFeedJobMessages` +- `setFeedJobErrStatus` +- `getPendingFeedJobScopes` +- `deleteDoneFeedJobs` + +`updateDeliveryJobStatus_` is duplicated as a feed-table statement rather than parameterised by table name. + +## 4. Worker + +`src/Simplex/Chat/Library/Subscriber.hs`: + +- `runDeliveryJobWorker` loses its `case deliveryKey` and handles groups only +- add `runFeedJobWorker` with the same `jobLoop`/`withWork_` shape +- extract the shared shape as `jobWorkerLoop :: Int64 -> Worker -> CM () -> CM ()` plus `jobOperation`, parameterised by the read, the processor and the error writer +- `getDeliveryJobWorker` handles `DeliveryWorkerKey`; add `getFeedJobWorker` for `FeedJobKey` +- `startDeliveryJobWorkers` reads group scopes; add `startFeedJobWorkers` reading feed scopes +- `startFeedWorkers` uses `getFeedJobWorker` + +## 5. Review items to resolve in the same change + +### Already resolved in the working tree + +- `DB.executeMany` with UPDATE and DELETE templates — replaced by `itemIdsStmt` (`IN ?` under `dbPostgres`, a loop otherwise) and by `forM_ … DB.execute`. The only remaining `executeMany` is the `INSERT` in `createFeedJobs`. +- `completeFeedJob` counts `job_status != DJSComplete`, so an errored sibling job prevents finalisation. +- `processFeedJob` sets `CISSndError` on the feed item when a creating action fails. +- The `CIDeleting` retry no longer downgrades to `FJADeleteInternal`; `createFeedJobs` supersedes errored jobs of the same action tag. +- The broadcast deletion window is enforced in the `CTFeed` branch as `diffUTCTime deletedTs itemTs < deleteMsgInterval`. +- `notification` derives from `feedActionEventTag action`. +- `messageDelete`, `groupMessageDelete` and the two file-description paths are silent under `chatDropsFeed`. +- `delGroupChatItems` is one `if` over `fullDelete`. +- `getChatItemIdsByAgentMsgId` is split into `getDirectChatItemIdsByAgentMsgId` and `getGroupChatItemIdsByAgentMsgId`, each probing both columns of the instance index. +- `getPendingDeliveryJobScopes` no longer drops rows whose scope fails to decode. +- The delivered-recipient filters use `Set` membership. +- The cursor and the first-bucket item status are written in one transaction. +- `ProtocolTests` covers `feed` on `x.msg.new` and `x.msg.update`. + +### To resolve + +1. **Errored feed job leaves the item unfinished.** `finishFeedEvent` never runs while a sibling is `DJSError`. Set the feed item to `CISSndError` and emit `CEvtChatItemsStatusesUpdated` when a job errors. Emit the same event for the first `SSPPartial` write. +2. **No retry of an errored feed job.** Decide between a bounded retry in the worker and a documented manual retry through the command. The current behaviour requires the user to repeat the command. +3. **A deleted contact aborts a whole bucket.** The `FJANew` bucket creates every instance in one `withStore'`, so one foreign key violation loses the rest of the broadcast. Use `withStoreBatch'` per recipient and drop the failing recipient. +4. **`finishFeedEvent` runs outside the feed lock.** Extend `withFeedLock` over `completeFeedJob` and `finishFeedEvent`. +5. **Group recipients.** Members that `memberSendAction` would treat as `MSAPending` or `MSAForwarded` receive nothing and get no pending message. Every member is recorded `GSSNew`, and one member's failure sets the whole instance to `CISSndError`. Record per-member `GSSError` as `createMemberSndStatuses` does, and decide whether pending members are stored. +6. **`APISendFeedMessage` is not atomic.** The item, the message, the message link and the jobs are four transactions. A crash between them leaves an undelivered item in `CISSndNew`. Create all four in one transaction. +7. **A locally marked feed item cannot be removed.** `(_, Just (CIDeleted _))` rejects every later deletion. Accept `CIDMInternal` on a marked item, as direct and group chats do. +8. **`dropFeed` suppression is too broad.** `XMsgDel` and `XMsgFileDescr` carry no feed marker, so the guards silence the not-found delete event and the missing-file error for every message in a chat with `dropFeed` set. Restrict the suppression to items whose shared id belongs to a dropped broadcast, or accept it and record the decision. +9. **Feed messages to groups are unsigned.** `createFeedMessage` passes no `MsgSigning` and `getFeedJobMessages` sets `signedMsg_ = Nothing`, so in a customer group with `SGFSignMessages` the broadcast arrives unverified. Decide between signing per group and excluding signing groups from the broadcast. +10. **`feeds.favorite` has no writer.** `APISetChatSettings` rejects `CTFeed`, and the favourite filter in `findFeedChatPreviews_` reads the column. Either accept `CTFeed` in `APISetChatSettings` or drop the column and the filter. +11. **`setUserChatsRead` skips `feeds`.** A feed marked unread stays unread after `APIUserRead`. +12. **`getChatRefViaItemId` rejects feed items.** `getAllChatItems` returns them, so `APIGetChatItems` after or around a feed item and `ShowChatItem` fail. Add `feed_id` to the query and the mapping. +13. **`SendFile` rejects `%`.** `APISendFeedMessage` accepts `fileSource`, and `/file %` does not reach it. +14. **`SEUserFeedNotFound` and `SEFeedNotFound` have no `viewStoreError` case.** +15. **`unzipMaybe3` unzips a pair.** Rename. +16. **`chat_query_plans.txt` is stale.** Regenerate it from a full test run. The file is owned by root in the current environment. +17. **`plans/2026-09-04-feed-broadcast-chat.md` contradicts the code.** Update it or drop it from the change. +18. **Tests.** Add coverage for `dropFeed`, a file broadcast, a restart in the middle of a job, incognito exclusion, a member send error, and a per-chat detach. `testFeedEditDelete` reads the instance immediately after the recipient prints the edit, before the worker writes it. +19. **`createContactPQSndItem` is not called on the feed path.** `deliverMessagesB` updates the stored PQ state, so only the chat item announcing the change is missing. +20. **Documentation.** A dropped broadcast still stores its body in `messages` for 30 days, and still returns a delivery receipt. + +## 6. Order of work + +1. Amend the migrations. +2. Split the types. +3. Move the store statements. +4. Split the worker. +5. Resolve items 1 to 9. +6. Resolve items 10 to 15. +7. Regenerate `chat_schema.sql`, `chat_lint.sql` and `chat_query_plans.txt`. +8. Add the tests of item 18. +9. Run the feed, group, direct, bot, schema and API-docs suites, and the Postgres target. diff --git a/src/Simplex/Chat/Delivery.hs b/src/Simplex/Chat/Delivery.hs index 6c893aa9d9..278a13be78 100644 --- a/src/Simplex/Chat/Delivery.hs +++ b/src/Simplex/Chat/Delivery.hs @@ -240,11 +240,14 @@ feedActionMsgIds = \case FJADeleteInternal -> [] FJADeleteMark -> [] -feedActionDelivers :: FeedJobAction -> Bool -feedActionDelivers = \case - FJADeleteInternal -> False - FJADeleteMark -> False - _ -> True +feedActionEventTag :: FeedJobAction -> CMEventTag 'Json +feedActionEventTag = \case + FJANew _ -> XMsgNew_ + FJAFileDescr _ -> XMsgFileDescr_ + FJAUpdate _ -> XMsgUpdate_ + FJADeleteBroadcast _ -> XMsgDel_ + FJADeleteInternal -> XMsgDel_ + FJADeleteMark -> XMsgDel_ feedActionCreates :: FeedJobAction -> Bool feedActionCreates = \case @@ -258,24 +261,25 @@ feedActionDeletes = \case FJADeleteMark -> True _ -> False --- a marked instance stays in its chat, so the feed item stays in the feed feedActionRemovesItem :: FeedJobAction -> Bool feedActionRemovesItem = \case FJADeleteBroadcast _ -> True FJADeleteInternal -> True _ -> False --- which instances of the feed item the action applies to data FeedInstanceSpec - = FISLinked -- a feed edit skips instances edited or deleted in their chat - | FISAny -- a feed deletion retracts the broadcast from detached instances too - | FISUndeleted -- a file description follows the message + = FISLinked + | FISAny + | FISUndeleted feedActionInstances :: FeedJobAction -> FeedInstanceSpec feedActionInstances = \case - FJAUpdate _ -> FISLinked + FJANew _ -> FISAny FJAFileDescr _ -> FISUndeleted - _ -> FISAny + FJAUpdate _ -> FISLinked + FJADeleteBroadcast _ -> FISAny + FJADeleteInternal -> FISAny + FJADeleteMark -> FISAny instance FromField FeedJobActionTag where fromField = fromTextField_ textDecode diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index 9137962732..e651237cf8 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -51,7 +51,7 @@ import Data.Text (Text) import qualified Data.Text as T import Data.Text.Encoding (decodeLatin1, encodeUtf8) import Data.Time (NominalDiffTime, addUTCTime, defaultTimeLocale, formatTime) -import Data.Time.Clock (UTCTime, getCurrentTime, nominalDay) +import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDay) import Data.Type.Equality import qualified Data.UUID as UUID import qualified Data.UUID.V4 as V4 @@ -818,7 +818,6 @@ processChatCommand cxt nm = \case when changed $ addInitialAndNewCIVersions db itemId (chatItemTs' ci, oldMC) (currentTs, mc) let edited = itemLive /= Just True - -- an edit in the chat detaches the instance: feed edits no longer apply to it when (itemFeed == Just CIFLinked) $ Store.detachFeedInstances db [itemId] updateDirectChatItem' db user contactId (detachedInstance ci) (CISndMsgContent mc) edited live Nothing $ Just msgId startUpdatedTimedItemThread user (ChatRef CTDirect contactId Nothing) ci ci' @@ -855,7 +854,6 @@ processChatCommand cxt nm = \case when changed $ addInitialAndNewCIVersions db itemId (chatItemTs' ci, oldMC) (currentTs, mc) let edited = itemLive /= Just True - -- an edit in the chat detaches the instance: feed edits no longer apply to it when (itemFeed == Just CIFLinked) $ Store.detachFeedInstances db [itemId] ci' <- updateGroupChatItem db user groupId (detachedInstance ci) (CISndMsgContent mc) edited live $ Just msgId updateGroupCIMentions db gInfo ci' ciMentions @@ -901,7 +899,6 @@ processChatCommand cxt nm = \case APIDeleteChatItem (ChatRef cType chatId scope) itemIds mode -> withUser $ \user -> case cType of CTDirect -> withContactLock "deleteChatItem" chatId $ do (ct, items) <- getCommandDirectChatItems user chatId itemIds - -- a deletion that keeps the row detaches the instance: feed edits no longer apply to it let markDeleted items' = do items'' <- detachFeedInstances items' markDirectCIsDeleted user ct items'' =<< liftIO getCurrentTime @@ -951,42 +948,35 @@ processChatCommand cxt nm = \case CTFeed -> withFeedLock "deleteChatItem" chatId $ do (feed, items) <- getCommandFeedChatItems user chatId itemIds when (null items) $ throwChatError CEInvalidChatItemDelete - case mode of - CIDMHistory -> throwChatError CEInvalidChatItemDelete - _ -> pure () deletions <- forM items $ \cci -> case cci of - CChatItem SMDSnd ci@ChatItem {meta = CIMeta {itemId, itemSharedMsgId, itemDeleted}} -> do - -- the item stays in the feed as CIDeleting until the jobs have removed every instance - action <- case (mode, itemDeleted) of - (_, Just (CIDeleted _)) -> throwChatError CEInvalidChatItemDelete - -- a repeated delete of an item being deleted re-enqueues the jobs with the earlier action - (_, Just (CIDeleting _)) -> feedDeleteAction feed itemId itemSharedMsgId CIDMInternal - (CIDMBroadcast, _) -> do - assertDeletable [cci] - feedDeleteAction feed itemId itemSharedMsgId CIDMBroadcast - (m, _) -> feedDeleteAction feed itemId itemSharedMsgId m + CChatItem SMDSnd ci@ChatItem {meta = CIMeta {itemId, itemSharedMsgId, itemTs, itemDeleted}} -> do deletedTs <- liftIO getCurrentTime + action <- case (mode, itemDeleted) of + (CIDMHistory, _) -> throwChatError CEInvalidChatItemDelete + (_, Just (CIDeleted _)) -> throwChatError CEInvalidChatItemDelete + (CIDMBroadcast, _) -> case itemSharedMsgId of + Just sharedMsgId | diffUTCTime deletedTs itemTs < deleteMsgInterval -> do + msg <- createFeedMessage feed Nothing itemId (XMsgDel sharedMsgId Nothing Nothing False) + pure $ FJADeleteBroadcast (msgId' msg) + _ -> throwChatError CEInvalidChatItemDelete + (CIDMInternalMark, Nothing) -> pure FJADeleteMark + _ -> pure FJADeleteInternal ci' <- withFastStore' $ \db -> do createFeedJobs db chatId itemId action - markFeedChatItemDeleted db user chatId ci (itemDeletedState mode deletedTs) deletedTs + markFeedChatItemDeleted db user chatId ci (feedItemDeleted action deletedTs) deletedTs pure $ ChatItemDeletion (aFeedItem feed cci) (Just $ AChatItem SCTFeed SMDSnd (FeedChat feed) ci') startFeedWorkers chatId pure $ CRChatItemsDeleted user deletions True False where - itemDeletedState m deletedTs = case m of - CIDMInternalMark -> CIDeleted (Just deletedTs) + feedItemDeleted a deletedTs = case a of + FJADeleteMark -> CIDeleted (Just deletedTs) _ -> CIDeleting (Just deletedTs) - feedDeleteAction feed itemId sharedMsgId_ = \case - CIDMBroadcast -> case sharedMsgId_ of - Nothing -> throwChatError CEInvalidChatItemDelete - Just sharedMsgId -> do - msg <- createFeedMessage feed Nothing itemId (XMsgDel sharedMsgId Nothing Nothing False) - pure $ FJADeleteBroadcast (msgId' msg) - CIDMInternalMark -> pure FJADeleteMark - _ -> pure FJADeleteInternal CTContactRequest -> throwCmdError "not supported" CTContactConnection -> throwCmdError "not supported" where + -- We check with a 6 hour margin compared to CIMeta deletable to account for deletion on the border + deleteMsgInterval :: NominalDiffTime + deleteMsgInterval = nominalDay + 6 * 3600 assertDeletable :: forall c. ChatTypeI c => [CChatItem c] -> CM () assertDeletable items = do currentTs <- liftIO getCurrentTime @@ -995,8 +985,7 @@ processChatCommand cxt nm = \case itemDeletable :: UTCTime -> CChatItem c -> Bool itemDeletable currentTs (CChatItem msgDir ChatItem {meta = CIMeta {itemSharedMsgId, itemTs, itemDeleted}, content}) = case msgDir of - -- We check with a 6 hour margin compared to CIMeta deletable to account for deletion on the border - SMDSnd -> isJust itemSharedMsgId && deletable' content itemDeleted itemTs (nominalDay + 6 * 3600) currentTs + SMDSnd -> isJust itemSharedMsgId && deletable' content itemDeleted itemTs deleteMsgInterval currentTs SMDRcv -> False itemsMsgIds :: [CChatItem c] -> [SharedMsgId] itemsMsgIds = mapMaybe (\(CChatItem _ ChatItem {meta = CIMeta {itemSharedMsgId}}) -> itemSharedMsgId) @@ -1223,7 +1212,6 @@ processChatCommand cxt nm = \case CTContactRequest -> throwCmdError "not supported" CTContactConnection -> throwCmdError "not supported" where - -- the user's own chats are not referenced as the source of a forward noRefComposeMsgReq :: CChatItem c -> (MsgContent, Maybe CryptoFile) -> ComposedMessageReq noRefComposeMsgReq (CChatItem _ ci) (mc', file) = let ciff = forwardCIFF ci Nothing @@ -1370,7 +1358,6 @@ processChatCommand cxt nm = \case user <- withFastStore $ \db -> getUserByNoteFolderId db chatId withFastStore' $ \db -> updateLocalChatItemsRead db user chatId ok user - -- the feed contains sent items only CTFeed -> ok =<< withFastStore (\db -> getUserByFeedId db chatId) CTContactRequest -> throwCmdError "not supported" CTContactConnection -> throwCmdError "not supported" @@ -1518,7 +1505,6 @@ processChatCommand cxt nm = \case withFastStore' $ \db -> deleteNoteFolderFiles db userId nf withFastStore' $ \db -> deleteNoteFolderCIs db user nf pure $ CRChatCleared user (AChatInfo SCTLocal $ LocalChat nf) - -- instances stay in their chats, with feed_item_id set to NULL by the FK CTFeed -> withFeedLock "clearChat" chatId $ do feed <- withFastStore $ \db -> getFeed db user chatId filesInfo <- withFastStore' $ \db -> getFeedFileInfo db user feed @@ -4253,14 +4239,11 @@ processChatCommand cxt nm = \case ciIds <- concat <$> withStore' (\db -> forM items $ \(CChatItem _ ci) -> markMessageReportsDeleted db user gInfo ci membership deletedTs) unless (null ciIds) $ toView $ CEvtGroupChatItemsDeleted user gInfo ciIds True (Just membership) let m = if moderation then Just membership else Nothing - -- a deletion that keeps the row detaches the instance: feed edits no longer apply to it - if groupFeatureUserAllowed SGFFullDelete gInfo && not moderation - then deleteGroupCIs user gInfo chatScopeInfo items m deletedTs - else do - items' <- detachFeedInstances items - if groupFeatureUserAllowed SGFFullDelete gInfo - then deleteGroupCIs user gInfo chatScopeInfo items' m deletedTs - else markGroupCIsDeleted user gInfo chatScopeInfo items' m deletedTs + fullDelete = groupFeatureUserAllowed SGFFullDelete gInfo + items' <- if fullDelete && not moderation then pure items else detachFeedInstances items + if fullDelete + then deleteGroupCIs user gInfo chatScopeInfo items' m deletedTs + else markGroupCIsDeleted user gInfo chatScopeInfo items' m deletedTs updateGroupProfileByName :: GroupName -> (GroupProfile -> GroupProfile) -> CM ChatResponse updateGroupProfileByName = updateGroupProfileByName_ Nothing updateGroupProfileByName_ :: Maybe GroupFeature -> GroupName -> (GroupProfile -> GroupProfile) -> CM ChatResponse @@ -5024,7 +5007,6 @@ processChatCommand cxt nm = \case withFastStore' $ \db -> createSndFTDescrXFTP db user (Just m) conn ft dummyFileDescr saveMemberFD _ = pure () - -- one description is sent to every feed recipient, so there are no snd_files rows CGFeed _ -> pure () pure (fInv, ciFile) prepareSndItemsData :: @@ -5855,7 +5837,6 @@ chatCommandP = ("\\\\ #" <|> "\\\\#") *> (DeleteMemberMessage <$> displayNameP <* A.space <* char_ '@' <*> displayNameP <* A.space <*> textP), ("! " <|> "!") *> (EditMessage <$> chatNameP <* A.space <*> (quotedMsg <|> pure "") <*> msgTextP), ReactToMessage <$> (("+" $> True) <|> ("-" $> False)) <*> reactionP <* A.space <*> chatNameP' <* A.space <*> textP, - -- "/feed drop" must precede "/feed ": msgTextP consumes any text "/feed drop " *> (SetDropFeed <$> chatNameP' <* A.space <*> onOffP), "/feed " *> (SendMessageBroadcast . MCText <$> msgTextP), ("/chats" <|> "/cs") *> (LastChats <$> (" all" $> Nothing <|> Just <$> (A.space *> A.decimal <|> pure 20))), diff --git a/src/Simplex/Chat/Library/Internal.hs b/src/Simplex/Chat/Library/Internal.hs index 739ecdb8fe..e6877df732 100644 --- a/src/Simplex/Chat/Library/Internal.hs +++ b/src/Simplex/Chat/Library/Internal.hs @@ -521,7 +521,6 @@ deleteFilesLocally files = withFilesFolder :: (FilePath -> CM ()) -> CM () withFilesFolder action = asks filesFolder >>= readTVarIO >>= mapM_ action --- a sender's feed instance renders the file of the feed item, which is not deleted with the instance itemsFilesInfo :: [CChatItem c] -> [CIFileInfo] itemsFilesInfo = mapMaybe itemFileInfo where @@ -529,7 +528,6 @@ itemsFilesInfo = mapMaybe itemFileInfo SMDSnd | isJust itemFeed -> Nothing _ -> mkCIFileInfo <$> file --- a per-chat edit or deletion detaches an instance: feed edits no longer apply to it detachFeedInstances :: forall c. [CChatItem c] -> CM [CChatItem c] detachFeedInstances items = do unless (null linkedIds) $ withStore' $ \db -> Store.detachFeedInstances db linkedIds @@ -2273,10 +2271,6 @@ createSndMessage chatMsgEvent connOrGroupId = createSndMessages :: forall e t. (MsgEncodingI e, Traversable t) => t (ConnOrGroupId, Maybe MsgSigning, ChatMsgEvent e) -> CM' (t (Either ChatError SndMessage)) createSndMessages = createSndMessages_ Nothing --- One message of a broadcast, linked to the feed item. The message of a new --- broadcast takes the id of the feed item, so that the item of every recipient --- has that id; the messages that follow take their own ids, as a group rejects --- a repeated message id (createNewRcvMessage, Store/Messages.hs:324). createFeedMessage :: Feed -> Maybe SharedMsgId -> ChatItemId -> ChatMsgEvent 'Json -> CM SndMessage createFeedMessage feed sharedMsgId_ feedItemId event = do msg <- liftEither . runIdentity =<< lift (createSndMessages_ sharedMsgId_ $ Identity (FeedId (feedId' feed), Nothing, event)) @@ -2662,7 +2656,6 @@ sendGroupSignedMessages_ gInfo@GroupInfo {groupId} recipientMembers signedEvents where mId = groupMemberId' m mIds' = S.insert mId mIds - -- the two batch modes share one deliverMessagesB call, so their body references do not overlap prepareMsgReqs :: MsgFlags -> NonEmpty (Either ChatError SndMessage) -> ([(GroupMember, Connection)], [(GroupMember, Connection)]) -> ([GroupMemberId], [Either ChatError ChatMsgReq]) prepareMsgReqs msgFlags msgs (toSendBin, toSendJson) = batchReqs 1 BMBinary toSendBin <> batchReqs 2 BMJson toSendJson @@ -2687,8 +2680,6 @@ sendGroupSignedMessages_ gInfo@GroupInfo {groupId} recipientMembers signedEvents createPendingMsg db (groupMemberId, msgId) = createPendingGroupMessage db groupMemberId msgId $> Right () --- The first recipient of a body sends it as a value, the rest reference it, --- so one encoded body is delivered to all of them. sharedBodyReqs :: forall r a. MsgFlags -> Int -> (Maybe Int -> Int -> a -> (ValueOrRef MsgBody, [MessageId])) -> NonEmpty (Either ChatError a) -> [(r, Connection)] -> ([r], [Either ChatError ChatMsgReq]) sharedBodyReqs msgFlags lastRef mkMb mbs recipients = snd $ foldr' foldMsgBodies (lastIdx_, ([], [])) recipients where diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index 1a16475aae..cdadcbb0b3 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -266,7 +266,6 @@ processAgentMsgSndFile _corrId aFileId msg = do where mConns' = mapMaybe readyMemberConn ms sfts' = mapMaybe (\sft@SndFileTransfer {groupMemberId} -> (,sft) <$> groupMemberId) sfts - -- one description is sent to every feed recipient by the feed jobs (rfd : _, _, SMDSnd, FeedChat feed@Feed {feedId}) -> do let feedItemId = fileItemId partSize <- asks $ xftpDescrPartSize . config @@ -1292,7 +1291,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = where updateGroupItemsErrorStatus :: DB.Connection -> AgentMsgId -> GroupMemberId -> GroupSndStatus -> IO () updateGroupItemsErrorStatus db msgId groupMemberId newStatus = do - itemIds <- getChatItemIdsByAgentMsgId db connId msgId + itemIds <- getGroupChatItemIdsByAgentMsgId db groupId connId msgId forM_ itemIds $ \itemId -> updateGroupMemSndStatus' db itemId groupMemberId newStatus getAutoReplyMsg = do let GroupInfo {businessChat} = gInfo @@ -1890,7 +1889,6 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = newContentMessage :: Contact -> MsgContainer -> RcvMessage -> MsgMeta -> CM () newContentMessage ct@Contact {chatSettings} mc msg@RcvMessage {sharedMsgId_} msgMeta - -- a feed message for a chat with dropFeed set is acknowledged and discarded | dropsFeedMsg chatSettings mc = pure () | otherwise = do let MsgContainer {content = c, file = fInv_} = mc @@ -1914,7 +1912,6 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = void $ newChatItem (ciContentNoParse $ CIRcvChatFeatureRejected CFVoice) Nothing Nothing False else do let MsgContainer {ttl = itemTTL, live = live_, feed = feed_} = mc - -- a feed message has no ttl in the body: the chat's own TTL applies itemTTL' = if feed_ == Just True then join (contactTimedTTL ct) else itemTTL timed_ = rcvContactCITimed ct itemTTL' live = fromMaybe False live_ @@ -1935,15 +1932,16 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = when (sz > fileSize) $ receiveFileEvt' user ft False Nothing Nothing >>= toView messageFileDescription :: Contact -> SharedMsgId -> FileDescr -> Maybe UTCTime -> CM () - messageFileDescription Contact {contactId} sharedMsgId fileDescr fileExpires = do - (fileId, aci) <- withStore $ \db -> do - fileId <- getFileIdBySharedMsgId db userId contactId sharedMsgId - aci <- getChatItemByFileId db cxt user fileId - pure (fileId, aci) - processFDMessage fileId aci fileDescr fileExpires + messageFileDescription Contact {contactId, chatSettings} sharedMsgId fileDescr fileExpires = + void $ skipDroppedFeedFile chatSettings $ do + (fileId, aci) <- withStore $ \db -> do + fileId <- getFileIdBySharedMsgId db userId contactId sharedMsgId + aci <- getChatItemByFileId db cxt user fileId + pure (fileId, aci) + processFDMessage fileId aci fileDescr fileExpires groupMessageFileDescription :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> FileDescr -> Maybe UTCTime -> CM (Maybe DeliveryTaskContext) - groupMessageFileDescription g@GroupInfo {groupId} m_ sharedMsgId fileDescr fileExpires = do + groupMessageFileDescription g@GroupInfo {groupId, chatSettings} m_ sharedMsgId fileDescr fileExpires = fmap join $ skipDroppedFeedFile chatSettings $ do (fileId, aci) <- withStore $ \db -> do fileId <- getGroupFileIdBySharedMsgId db userId groupId sharedMsgId aci <- getChatItemByFileId db cxt user fileId @@ -2005,7 +2003,6 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = messageUpdate :: Contact -> SharedMsgId -> MsgContent -> RcvMessage -> MsgMeta -> Maybe Int -> Maybe Bool -> Maybe Bool -> CM () messageUpdate ct@Contact {contactId, chatSettings} sharedMsgId mc msg@RcvMessage {msgId} msgMeta ttl live_ feed_ = do updateRcvChatItem `catchCINotFound` \_ -> - -- an update of a dropped feed message creates nothing unless (dropsFeed chatSettings feed_) $ do -- This patches initial sharedMsgId into chat item when locally deleted chat item -- received an update from the sender, so that it can be referenced later (e.g. by broadcast delete). @@ -2048,8 +2045,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = _ -> messageError "x.msg.update: contact attempted invalid message update" messageDelete :: Contact -> SharedMsgId -> RcvMessage -> MsgMeta -> CM () - messageDelete ct@Contact {contactId} sharedMsgId _rcvMessage msgMeta = do - deleteRcvChatItem `catchCINotFound` (toView . CEvtChatItemDeletedNotFound user ct) + messageDelete ct@Contact {contactId, chatSettings} sharedMsgId _rcvMessage msgMeta = do + deleteRcvChatItem `catchCINotFound` \smId -> + unless (chatDropsFeed chatSettings) $ toView $ CEvtChatItemDeletedNotFound user ct smId where brokerTs = metaBrokerTs msgMeta deleteRcvChatItem = do @@ -2138,6 +2136,12 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = ChatErrorStore (SEChatItemSharedMsgIdNotFound sharedMsgId) -> handle sharedMsgId e -> throwError e + skipDroppedFeedFile :: ChatSettings -> CM a -> CM (Maybe a) + skipDroppedFeedFile chatSettings f = + (Just <$> f) `catchAllErrors` \case + ChatErrorStore (SEFileIdNotFoundBySharedMsgId _) | chatDropsFeed chatSettings -> pure Nothing + e -> throwError e + validSender :: Maybe GroupMember -> CIDirection 'CTGroup 'MDRcv -> Bool validSender (Just m) (CIGroupRcv mem) = sameMemberId (memberId' m) mem validSender m_ CIChannelRcv = maybe True (\m -> memberRole' m == GROwner) m_ @@ -2149,7 +2153,6 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = newGroupContentMessage :: GroupInfo -> Maybe GroupMember -> MsgContainer -> RcvMessage -> UTCTime -> Bool -> CM (Maybe DeliveryTaskContext) newGroupContentMessage gInfo@GroupInfo {chatSettings} m_ mc msg@RcvMessage {sharedMsgId_} brokerTs forwarded - -- a feed message for a chat with dropFeed set is acknowledged and discarded | dropsFeedMsg chatSettings mc = pure Nothing | otherwise = case m_ of Nothing -> do @@ -2177,7 +2180,6 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = pure $ Just $ infoToDeliveryContext gInfo' scopeInfo sentAsGroup where rejected gInfo' m' scopeInfo f = newChatItem gInfo' m' scopeInfo (ciContentNoParse $ CIRcvGroupFeatureRejected f) Nothing Nothing False - -- a feed message has no ttl in the body: the chat's own TTL applies itemTTL' gInfo' = if feed_ == Just True then join (groupTimedTTL gInfo') else itemTTL timed_ gInfo' = if forwarded then rcvCITimed_ (Just Nothing) (itemTTL' gInfo') else rcvGroupCITimed gInfo' (itemTTL' gInfo') live' = fromMaybe False live_ @@ -2335,7 +2337,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = pure Nothing groupMessageDelete :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> Maybe MemberId -> Maybe MsgScope -> Bool -> RcvMessage -> UTCTime -> CM (Maybe DeliveryTaskContext) - groupMessageDelete gInfo@GroupInfo {membership} m_ sharedMsgId sndMemberId_ scope_ onlyHistory rcvMsg brokerTs = + groupMessageDelete gInfo@GroupInfo {membership, chatSettings = groupChatSettings} m_ sharedMsgId sndMemberId_ scope_ onlyHistory rcvMsg brokerTs = findItem >>= \case Right cci@(CChatItem _ ci@ChatItem {chatDir}) -> requireVerifiedDelete cci $ case (chatDir, m_) of (CIGroupRcv mem, Just m@GroupMember {memberId}) -> @@ -2366,7 +2368,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = let msgMemberId = fromMaybe memberId sndMemberId_ if | msgMemberId == memberId -> - messageError ("x.msg.del: message not found, " <> tshow e) $> Nothing + unless (chatDropsFeed groupChatSettings) (messageError $ "x.msg.del: message not found, " <> tshow e) $> Nothing | senderRole < GRModerator -> do messageError $ "x.msg.del: message not found, message of another member with insufficient member permissions, " <> tshow e pure Nothing @@ -4153,7 +4155,6 @@ getDeliveryJobWorker hasWork deliveryKey = do getAgentWorker "delivery_job" hasWork a deliveryKey ws $ runDeliveryJobWorker a deliveryKey --- both feed streams of the event are started: each has its own worker startFeedWorkers :: FeedId -> CM () startFeedWorkers feedId = lift $ mapM_ (\scope -> void $ getDeliveryJobWorker True (DJKFeed feedId scope)) feedWorkerScopes @@ -4392,13 +4393,14 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do bucketSize <- asks $ feedBucketSize . config let msgIds = feedActionMsgIds feedAction msgs <- withStore' $ \db -> getFeedJobMessages db msgIds - -- a job pending longer than the message retention loses its body: the local changes still apply when (length msgs < length msgIds) $ logWarn "feed job: message not found" - bucketLoop bucketSize msgs startingCursor + bucketLoop bucketSize msgs startingCursor `catchAllErrors` \e -> do + when (feedActionCreates feedAction) $ + withStore' $ \db -> updateFeedChatItemStatus db user feedId feedItemId (CISSndError $ SndErrOther $ tshow e) + throwError e where DeliveryJob {jobId, cursorId_ = startingCursor, jobWork = DJWFeed {feedItemId, feedAction}} = job Feed {feedId} = feed - -- the feed item is re-read for each bucket: a deletion issued while the job runs stops it bucketLoop :: Int -> [SndMessage] -> Maybe Int64 -> CM () bucketLoop bucketSize msgs cursor_ = withStore' (\db -> runExceptT $ getFeedChatItem db user feedId feedItemId) >>= \case @@ -4415,10 +4417,10 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do cursor' <- case scope of FWSContacts -> feedContactsBucket cxt user feedItemId feedAction item bucketSize msgs cursor_ FWSGroups -> feedGroupsBucket cxt user feedItemId feedAction item bucketSize msgs cursor_ - forM_ cursor' $ \c -> do - withStore' $ \db -> updateFeedDeliveryJobCursor db jobId c + forM_ cursor' $ \c -> withStore' $ \db -> do + updateFeedDeliveryJobCursor db jobId c when (feedActionCreates feedAction && isNothing cursor_) $ - withStore' $ \db -> updateFeedChatItemStatus db user feedId feedItemId (CISSndSent SSPPartial) + updateFeedChatItemStatus db user feedId feedItemId (CISSndSent SSPPartial) pure cursor' completeJob = do lastOfEvent <- @@ -4426,20 +4428,19 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do withStore' $ \db -> completeFeedJob db jobId feedItemId (feedActionTag feedAction) when lastOfEvent $ finishFeedEvent user feed feedItemId feedAction --- a recipient with dropFeed set discards feed messages of the chat quietly +chatDropsFeed :: ChatSettings -> Bool +chatDropsFeed ChatSettings {dropFeed} = isTrue dropFeed + dropsFeed :: ChatSettings -> Maybe Bool -> Bool -dropsFeed ChatSettings {dropFeed} feed_ = feed_ == Just True && isTrue dropFeed +dropsFeed chatSettings feed_ = feed_ == Just True && chatDropsFeed chatSettings dropsFeedMsg :: ChatSettings -> MsgContainer -> Bool dropsFeedMsg chatSettings MsgContainer {feed} = dropsFeed chatSettings feed --- the delete time of a sent instance, from the chat's disappearing messages preference sndFeedTimed :: Maybe (Maybe Int) -> UTCTime -> Maybe CITimed sndFeedTimed chatTTL createdAt = (\ttl -> CITimed ttl (Just $ addUTCTime (realToFrac ttl) createdAt)) <$> join chatTTL --- One bucket of contacts: instances are created or changed, then the bodies are --- delivered, then the caller writes the cursor. Nothing ends the stream. feedContactsBucket :: StoreCxt -> User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> Int -> [SndMessage] -> Maybe ContactId -> CM (Maybe ContactId) feedContactsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case action of FJANew _ -> do @@ -4465,7 +4466,7 @@ feedContactsBucket cxt user feedItemId action item bucketSize msgs cursor_ = cas _ -> do instances <- withStore' $ \db -> getFeedContactInstancesByCursor db cxt user feedItemId (feedActionInstances action) cursor_ bucketSize let ctIds = map (contactId' . fst) instances - when (feedActionDelivers action) $ + unless (null msgs) $ forM_ (lastId ctIds) $ \lastCtId -> void $ deliverBucket lastCtId [(contactId' ct, itemId, conn) | (ct, itemId) <- instances, Just (_, conn) <- [feedRecipient ct]] applyContactAction user feedItemId action item instances fromId (fromMaybe 0 $ lastId ctIds) @@ -4476,16 +4477,13 @@ feedContactsBucket cxt user feedItemId action item bucketSize msgs cursor_ = cas feedRecipient ct = case contactSendConn_ ct of Right conn | directOrUsed ct && not (connIncognito conn) -> Just (ct, conn) _ -> Nothing - -- a resumed job skips the connections the agent accepted the message for deliverBucket lastCtId recipients = do - delivered <- withStore' $ \db -> getDeliveredContactIdsByRange db (firstMsgId msgs) fromId lastCtId - deliverFeedBucket msgs [(itemId, conn) | (ctId, itemId, conn) <- recipients, ctId `notElem` delivered] + delivered <- withStore' $ \db -> S.fromList <$> getDeliveredContactIdsByRange db (firstMsgId msgs) fromId lastCtId + deliverFeedBucket action msgs [(itemId, conn) | (ctId, itemId, conn) <- recipients, not $ ctId `S.member` delivered] --- instance changes of one contacts bucket, in one transaction applyContactAction :: User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> [(Contact, ChatItemId)] -> ContactId -> ContactId -> CM () applyContactAction user feedItemId action item instances fromId toId = case action of FJAUpdate _ -> withStore' $ \db -> updateFeedContactInstances db user feedItemId fromId toId item - -- each contact's own full delete preference decides between deleting and marking FJADeleteBroadcast _ -> do let (toDelete, toMark) = partition (fullDelete . fst) instances deleteItems toDelete @@ -4503,7 +4501,6 @@ applyContactAction user feedItemId action item instances fromId toId = case acti deletedTs <- liftIO getCurrentTime withStore' $ \db -> markFeedInstancesDeleted db (map snd cts) deletedTs --- One bucket of customer groups. feedGroupsBucket :: StoreCxt -> User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> Int -> [SndMessage] -> Maybe GroupId -> CM (Maybe GroupId) feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case action of FJANew _ -> do @@ -4521,7 +4518,6 @@ feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case Just itemId -> pure (groupId' g, itemId, ms, Nothing) Nothing -> do itemId <- createFeedInstanceItem db user (CDGroupSnd g Nothing) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt - -- member statuses of a new instance, so that SENT and receipts of its members are recorded forM_ ms $ \(m, _) -> createGroupSndStatus db itemId (groupMemberId' m) GSSNew pure (groupId' g, itemId, ms, timed_) forM_ recipients $ \(gId, itemId, _, newTimed_) -> @@ -4533,7 +4529,7 @@ feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case _ -> do instances <- withStore' $ \db -> getFeedGroupInstancesByCursor db cxt user feedItemId (feedActionInstances action) cursor_ bucketSize let gIds = map (groupId' . fst) instances - when (feedActionDelivers action) $ + unless (null msgs) $ forM_ (lastId gIds) $ \lastGId -> do members <- withStore' $ \db -> getCustomerGroupsMembersByRange db cxt user fromId lastGId void $ deliverBucket lastGId [(itemId, groupMembers members g) | (g, itemId) <- instances] @@ -4546,15 +4542,13 @@ feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case not (incognitoMembership g) && memberCurrent membership && memberActive membership groupMembers members g = [(m, conn) | m <- filter memberCurrent (M.findWithDefault [] (groupId' g) members), Just (_, conn) <- [readyMemberConn m]] - -- a resumed job skips the connections the agent accepted the message for deliverBucket lastGId recipients = do - delivered <- withStore' $ \db -> getDeliveredMemberIdsByRange db (firstMsgId msgs) fromId lastGId - deliverFeedBucket msgs [(itemId, conn) | (itemId, ms) <- recipients, (m, conn) <- ms, groupMemberId' m `notElem` delivered] + delivered <- withStore' $ \db -> S.fromList <$> getDeliveredMemberIdsByRange db (firstMsgId msgs) fromId lastGId + deliverFeedBucket action msgs [(itemId, conn) | (itemId, ms) <- recipients, (m, conn) <- ms, not $ groupMemberId' m `S.member` delivered] applyGroupAction :: User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> [(GroupInfo, ChatItemId)] -> GroupId -> GroupId -> CM () applyGroupAction user feedItemId action item instances fromId toId = case action of FJAUpdate _ -> withStore' $ \db -> updateFeedGroupInstances db user feedItemId fromId toId item - -- each group's own full delete preference decides between deleting and marking FJADeleteBroadcast _ -> do let (toDelete, toMark) = partition (groupFeatureUserAllowed SGFFullDelete . fst) instances deleteItems toDelete @@ -4572,15 +4566,14 @@ applyGroupAction user feedItemId action item instances fromId toId = case action withStore' $ \db -> markFeedInstancesDeleted db (map snd gs) deletedTs membershipId GroupInfo {groupId, membership} = (groupId, memberId' membership) --- one deliverMessagesB for the bucket: the first request of each body sends it, the rest reference it -deliverFeedBucket :: [SndMessage] -> [(ChatItemId, Connection)] -> CM [(ChatItemId, Either ChatError ([Int64], PQEncryption))] -deliverFeedBucket msgs recipients = case (L.nonEmpty msgs, recipients) of +deliverFeedBucket :: FeedJobAction -> [SndMessage] -> [(ChatItemId, Connection)] -> CM [(ChatItemId, Either ChatError ([Int64], PQEncryption))] +deliverFeedBucket action msgs recipients = case (L.nonEmpty msgs, recipients) of (Just msgs', _ : _) -> do let batched_ = batchSndMessagesJSON BMJson $ L.map Right msgs' case L.nonEmpty batched_ of Nothing -> pure [] Just batched' -> do - let msgFlags = MsgFlags {notification = hasNotification XMsgNew_} + let msgFlags = MsgFlags {notification = hasNotification $ feedActionEventTag action} (itemIds, msgReqs) = sharedBodyReqs msgFlags (length batched' + length msgs') msgBatchMBR batched' recipients case L.nonEmpty msgReqs of Nothing -> pure [] @@ -4604,13 +4597,11 @@ lastId ids = case reverse ids of i : _ -> Just i [] -> Nothing --- the stream continues while buckets are full bucketCursor :: Int -> [Int64] -> Maybe Int64 bucketCursor bucketSize ids | length ids < bucketSize = Nothing | otherwise = lastId ids --- the last job of the event reports its result on the feed item finishFeedEvent :: User -> Feed -> ChatItemId -> FeedJobAction -> CM () finishFeedEvent user feed@Feed {feedId} feedItemId action | feedActionCreates action = do @@ -4626,7 +4617,6 @@ finishFeedEvent user feed@Feed {feedId} feedItemId action withStore' (\db -> runExceptT $ getFeedChatItem db user feedId feedItemId) >>= either (const $ pure ()) action' - -- Single worker processes all relay requests (XGrpRelayInv). -- We use map with a single key 1 to fit into existing worker management framework. relayRequestWorkerKey :: Int diff --git a/src/Simplex/Chat/Store/Delivery.hs b/src/Simplex/Chat/Store/Delivery.hs index a0ee6df355..a4eaacfb57 100644 --- a/src/Simplex/Chat/Store/Delivery.hs +++ b/src/Simplex/Chat/Store/Delivery.hs @@ -39,7 +39,7 @@ import Data.ByteString.Char8 (ByteString) import Data.Int (Int64) import qualified Data.List.NonEmpty as L import Data.List.NonEmpty (NonEmpty (..)) -import Data.Maybe (catMaybes, isNothing, mapMaybe) +import Data.Maybe (catMaybes, isNothing) import Data.Text (Text) import qualified Data.Text as T import Data.Time.Clock (UTCTime, getCurrentTime) @@ -53,7 +53,6 @@ import Simplex.Messaging.Agent.Store.AgentStore (getWorkItem, getWorkItems, mayb import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..)) import qualified Simplex.Messaging.Agent.Store.DB as DB import Simplex.Messaging.Encoding (smpDecode) -import Simplex.Messaging.Encoding.String (TextEncoding (..)) import Simplex.Messaging.Util (eitherToMaybe, firstRow') import Text.Read (readMaybe) #if defined(dbPostgres) @@ -278,22 +277,26 @@ createMsgDeliveryJob db gInfo jobScope senderGMIds body = do | otherwise = Just $ T.intercalate "," $ map (T.pack . show) senderGMIds getPendingDeliveryJobScopes :: DB.Connection -> IO [DeliveryJobKey] -getPendingDeliveryJobScopes db = - mapMaybe toJobKey - <$> DB.query +getPendingDeliveryJobScopes db = do + groupScopes <- + DB.query db [sql| - SELECT DISTINCT group_id, feed_id, worker_scope + SELECT DISTINCT group_id, worker_scope FROM delivery_jobs - WHERE failed = 0 AND job_status = ? + WHERE failed = 0 AND job_status = ? AND group_id IS NOT NULL |] (Only DJSPending) - where - toJobKey :: (Maybe GroupId, Maybe FeedId, Text) -> Maybe DeliveryJobKey - toJobKey = \case - (Just groupId, Nothing, scope) -> DJKGroup groupId <$> (textDecode scope :: Maybe DeliveryWorkerScope) - (Nothing, Just feedId, scope) -> DJKFeed feedId <$> (textDecode scope :: Maybe FeedWorkerScope) - _ -> Nothing + feedScopes <- + DB.query + db + [sql| + SELECT DISTINCT feed_id, worker_scope + FROM delivery_jobs + WHERE failed = 0 AND job_status = ? AND feed_id IS NOT NULL + |] + (Only DJSPending) + pure $ map (uncurry DJKGroup) groupScopes <> map (uncurry DJKFeed) feedScopes type MessageDeliveryJobRow = (Only Int64) :. DeliveryJobScopeRow :. (Maybe Text, Binary ByteString, Maybe GroupMemberId) @@ -337,9 +340,6 @@ getNextDeliveryJob db deliveryKey = do senderGMIds <- parseIds jobId' senderGMIdsText_ Right DeliveryJob {jobId = jobId', cursorId_, jobWork = DJWGroup {jobScope, senderGMIds, body}} --- NULL or empty string means []; otherwise the value must parse as a --- comma-separated decimal Int64 list. An unparseable segment surfaces as --- job error rather than silent degradation. parseIds :: Int64 -> Maybe Text -> Either StoreError [Int64] parseIds jobId = \case Nothing -> Right [] @@ -356,10 +356,14 @@ markJobFailed :: DB.Connection -> Int64 -> IO () markJobFailed db jobId = DB.execute db "UPDATE delivery_jobs SET failed = 1 where delivery_job_id = ?" (Only jobId) -createFeedDeliveryJob :: DB.Connection -> FeedId -> ChatItemId -> FeedWorkerScope -> FeedJobAction -> IO () -createFeedDeliveryJob db feedId feedItemId scope action = do +createFeedJobs :: DB.Connection -> FeedId -> ChatItemId -> FeedJobAction -> IO () +createFeedJobs db feedId feedItemId action = do currentTs <- getCurrentTime DB.execute + db + "DELETE FROM delivery_jobs WHERE chat_item_id = ? AND job_scope_spec_tag = ? AND (job_status = ? OR failed = 1)" + (feedItemId, tag, DJSError) + DB.executeMany db [sql| INSERT INTO delivery_jobs ( @@ -367,11 +371,10 @@ createFeedDeliveryJob db feedId feedItemId scope action = do job_status, created_at, updated_at ) VALUES (?,?,?,?,?,?,?,?) |] - (feedId, feedItemId, scope, feedActionTag action, idsColumn (feedActionMsgIds action), DJSPending, currentTs, currentTs) - -createFeedJobs :: DB.Connection -> FeedId -> ChatItemId -> FeedJobAction -> IO () -createFeedJobs db feedId feedItemId action = - mapM_ (\scope -> createFeedDeliveryJob db feedId feedItemId scope action) feedWorkerScopes + [(feedId, feedItemId, scope, tag, msgIds, DJSPending, currentTs, currentTs) | scope <- feedWorkerScopes] + where + tag = feedActionTag action + msgIds = idsColumn $ feedActionMsgIds action type FeedDeliveryJobRow = (Int64, ChatItemId, FeedJobActionTag, Maybe Text, Maybe Int64) @@ -425,23 +428,21 @@ updateFeedDeliveryJobCursor db jobId cursorId = do "UPDATE delivery_jobs SET feed_cursor_id = ?, updated_at = ? WHERE delivery_job_id = ?" (cursorId, currentTs, jobId) --- completes the job and reports whether it was the last job of its feed event completeFeedJob :: DB.Connection -> Int64 -> ChatItemId -> FeedJobActionTag -> IO Bool completeFeedJob db jobId feedItemId actionTag = do updateDeliveryJobStatus db jobId DJSComplete - pending <- + unfinished <- maybeFirstRow fromOnly $ DB.query db [sql| SELECT COUNT(1) FROM delivery_jobs - WHERE chat_item_id = ? AND job_scope_spec_tag = ? AND job_status = ? AND failed = 0 + WHERE chat_item_id = ? AND job_scope_spec_tag = ? AND job_status != ? |] - (feedItemId, actionTag, DJSPending) - pure $ pending == Just (0 :: Int) + (feedItemId, actionTag, DJSComplete) + pure $ unfinished == Just (0 :: Int) --- the messages of a feed job action: one, or the parts of one file description getFeedJobMessages :: DB.Connection -> [MessageId] -> IO [SndMessage] getFeedJobMessages db = fmap catMaybes . mapM getMsg where diff --git a/src/Simplex/Chat/Store/Feeds.hs b/src/Simplex/Chat/Store/Feeds.hs index 11d4ada4f8..d728b56282 100644 --- a/src/Simplex/Chat/Store/Feeds.hs +++ b/src/Simplex/Chat/Store/Feeds.hs @@ -38,6 +38,7 @@ module Simplex.Chat.Store.Feeds ) where +import Control.Monad (forM_, unless) import Control.Monad.Except (ExceptT (..), throwError) import Control.Monad.IO.Class (liftIO) import Data.Int (Int64) @@ -55,10 +56,10 @@ import Simplex.Messaging.Agent.Store.AgentStore (firstRow) import Simplex.Messaging.Agent.Store.DB (BoolInt (..)) import qualified Simplex.Messaging.Agent.Store.DB as DB #if defined(dbPostgres) -import Database.PostgreSQL.Simple (Only (..), Query, (:.) (..)) +import Database.PostgreSQL.Simple (In (..), Only (..), Query, ToRow, (:.) (..)) import Database.PostgreSQL.Simple.SqlQQ (sql) #else -import Database.SQLite.Simple (Only (..), Query, (:.) (..)) +import Database.SQLite.Simple (Only (..), Query, ToRow, (:.) (..)) import Database.SQLite.Simple.QQ (sql) #endif @@ -93,13 +94,11 @@ updateFeedUnreadChat db User {userId} Feed {feedId} unreadChat = do updatedAt <- getCurrentTime DB.execute db "UPDATE feeds SET unread_chat = ?, updated_at = ? WHERE user_id = ? AND feed_id = ?" (BI unreadChat, updatedAt, userId, feedId) --- messages are deleted first: chat_item_messages rows cascade with chat items deleteFeedCIs :: DB.Connection -> User -> Feed -> IO () deleteFeedCIs db User {userId} Feed {feedId} = do DB.execute db "DELETE FROM messages WHERE feed_id = ?" (Only feedId) DB.execute db "DELETE FROM chat_items WHERE user_id = ? AND feed_id = ?" (userId, feedId) --- the chats of a bucket are read without their chat tags, which a job does not use getFeedContactsByCursor :: DB.Connection -> StoreCxt -> User -> Maybe ContactId -> Int -> IO [Contact] getFeedContactsByCursor db cxt user@User {userId} cursorId_ count = do currentTs <- getCurrentTime @@ -118,7 +117,6 @@ getFeedCustomerGroupsByCursor db cxt User {userId, userContactId} cursorId_ coun (groupInfoQuery <> " WHERE g.user_id = ? AND mu.contact_id = ? AND g.business_chat = ? AND g.group_id > ? ORDER BY g.group_id ASC LIMIT ?") (userId, userContactId, BCCustomer, cursorId cursorId_, count) --- members of the user's customer groups in the group id range, without the user's own membership getCustomerGroupsMembersByRange :: DB.Connection -> StoreCxt -> User -> GroupId -> GroupId -> IO (Map GroupId [GroupMember]) getCustomerGroupsMembersByRange db cxt user@User {userId, userContactId} fromId toId = do currentTs <- getCurrentTime @@ -138,7 +136,6 @@ getCustomerGroupsMembersByRange db cxt user@User {userId, userContactId} fromId foldMembers = foldr (\m -> M.insertWith (<>) (memberGroupId m) [m]) M.empty memberGroupId GroupMember {groupId} = groupId --- the part of a feed item that is repeated in its instances data FeedItemMsg = FeedItemMsg { feedSharedMsgId :: SharedMsgId, feedContent :: CIContent 'MDSnd, @@ -193,7 +190,6 @@ getFeedGroupInstancesByCursor db cxt User {userId, userContactId} feedItemId spe ) (userId, userContactId, feedItemId, cursorId cursorId_, count) --- instances created by an earlier run of the same bucket, by chat getFeedInstanceContactIdsByRange :: DB.Connection -> User -> ChatItemId -> ContactId -> ContactId -> IO (Map ContactId ChatItemId) getFeedInstanceContactIdsByRange db user feedItemId = getFeedInstanceIdsByRange_ db user feedItemId "contact_id" @@ -214,12 +210,12 @@ getFeedInstanceIdsByRange_ db User {userId} feedItemId chatIdColumn fromId toId updateFeedInstanceStatuses :: DB.Connection -> [(ChatItemId, CIStatus 'MDSnd)] -> IO () updateFeedInstanceStatuses db statuses = do currentTs <- getCurrentTime - DB.executeMany - db - "UPDATE chat_items SET item_status = ?, updated_at = ? WHERE chat_item_id = ?" - (map (\(itemId, status) -> (status, currentTs, itemId)) statuses) + forM_ statuses $ \(itemId, status) -> + DB.execute + db + "UPDATE chat_items SET item_status = ?, updated_at = ? WHERE chat_item_id = ?" + (status, currentTs, itemId) --- a msg_deliveries row exists once the agent accepted the message for the connection getDeliveredContactIdsByRange :: DB.Connection -> MessageId -> ContactId -> ContactId -> IO [ContactId] getDeliveredContactIdsByRange db msgId fromId toId = map fromOnly @@ -266,36 +262,43 @@ updateFeedInstances_ db User {userId} feedItemId chatIdColumn fromId toId FeedIt deleteFeedInstances :: DB.Connection -> [ChatItemId] -> IO () deleteFeedInstances db itemIds = do - DB.executeMany db "DELETE FROM chat_item_messages WHERE chat_item_id = ?" (map Only itemIds) - DB.executeMany db "DELETE FROM chat_item_versions WHERE chat_item_id = ?" (map Only itemIds) - DB.executeMany db "DELETE FROM chat_items WHERE chat_item_id = ?" (map Only itemIds) + itemIdsStmt db "DELETE FROM chat_item_messages WHERE chat_item_id" () itemIds + itemIdsStmt db "DELETE FROM chat_item_versions WHERE chat_item_id" () itemIds + itemIdsStmt db "DELETE FROM chat_items WHERE chat_item_id" () itemIds markFeedInstancesDeleted :: DB.Connection -> [ChatItemId] -> UTCTime -> IO () markFeedInstancesDeleted db itemIds deletedTs = do currentTs <- getCurrentTime - DB.executeMany + itemIdsStmt db - "UPDATE chat_items SET item_deleted = 1, item_deleted_ts = ?, updated_at = ? WHERE chat_item_id = ?" - (map (deletedTs,currentTs,) itemIds) + "UPDATE chat_items SET item_deleted = 1, item_deleted_ts = ?, updated_at = ? WHERE chat_item_id" + (deletedTs, currentTs) + itemIds detachFeedInstances :: DB.Connection -> [ChatItemId] -> IO () detachFeedInstances db itemIds = - DB.executeMany db "UPDATE chat_items SET item_feed = 2 WHERE chat_item_id = ? AND item_feed = 1" (map Only itemIds) + itemIdsStmt db "UPDATE chat_items SET item_feed = 2 WHERE item_feed = 1 AND chat_item_id" () itemIds deleteFeedContactReactions :: DB.Connection -> SharedMsgId -> [ContactId] -> IO () deleteFeedContactReactions db sharedMsgId contactIds = - DB.executeMany - db - "DELETE FROM chat_item_reactions WHERE contact_id = ? AND shared_msg_id = ?" - (map (,sharedMsgId) contactIds) + itemIdsStmt db "DELETE FROM chat_item_reactions WHERE shared_msg_id = ? AND contact_id" (Only sharedMsgId) contactIds --- reactions to the user's own group items are stored with the membership member id deleteFeedGroupReactions :: DB.Connection -> SharedMsgId -> [(GroupId, MemberId)] -> IO () deleteFeedGroupReactions db sharedMsgId groupMemberIds = - DB.executeMany - db - "DELETE FROM chat_item_reactions WHERE group_id = ? AND shared_msg_id = ? AND item_member_id = ?" - (map (\(gId, memId) -> (gId, sharedMsgId, memId)) groupMemberIds) + forM_ groupMemberIds $ \(groupId, memberId) -> + DB.execute + db + "DELETE FROM chat_item_reactions WHERE group_id = ? AND shared_msg_id = ? AND item_member_id = ?" + (groupId, sharedMsgId, memberId) + +itemIdsStmt :: ToRow p => DB.Connection -> Query -> p -> [Int64] -> IO () +itemIdsStmt db stmt params itemIds = unless (null itemIds) execIds + where +#if defined(dbPostgres) + execIds = DB.execute db (stmt <> " IN ?") (params :. Only (In itemIds)) +#else + execIds = forM_ itemIds $ \itemId -> DB.execute db (stmt <> " = ?") (params :. Only itemId) +#endif cursorId :: Maybe Int64 -> Int64 cursorId = fromMaybe 0 diff --git a/src/Simplex/Chat/Store/Messages.hs b/src/Simplex/Chat/Store/Messages.hs index 7346692d8f..6652ed7973 100644 --- a/src/Simplex/Chat/Store/Messages.hs +++ b/src/Simplex/Chat/Store/Messages.hs @@ -106,7 +106,7 @@ module Simplex.Chat.Store.Messages getGroupReactions, setGroupReaction, getReactionMembers, - getChatItemIdsByAgentMsgId, + getGroupChatItemIdsByAgentMsgId, getDirectChatItem, getDirectCIWithReactions, getDirectChatItemBySharedMsgId, @@ -328,7 +328,6 @@ createNewRcvMessage db connOrGroupId NewRcvMessage {chatMsgEvent, verifiedMsg, b throwError $ SEDuplicateGroupMessage groupId sharedMsgId duplAuthorId duplFwdMemberId Nothing -> liftIO $ insertRcvMsg Nothing $ Just groupId Nothing -> liftIO $ insertRcvMsg Nothing $ Just groupId - -- received messages arrive on a connection, feed messages are only sent FeedId _ -> throwError $ SEInternalError "received message with feed entity" where duplicateGroupMsgMemberIds :: Int64 -> SharedMsgId -> IO (Maybe (Maybe GroupMemberId, Maybe GroupMemberId)) @@ -609,7 +608,6 @@ createNewChatItemNoMsg db user chatDirection showGroupAsSender ciContent sharedM quoteRow :: NewQuoteRow quoteRow = (Nothing, Nothing, Nothing, Nothing, Nothing) --- an instance of a feed item in a contact or group chat: no message id, the feed item's shared id createFeedInstanceItem :: DB.Connection -> User -> ChatDirection c 'MDSnd -> SharedMsgId -> CIContent 'MDSnd -> ChatItemId -> Maybe CITimed -> Bool -> UTCTime -> IO ChatItemId createFeedInstanceItem db user chatDirection sharedMsgId ciContent feedItemId timed hasLink createdAt = createNewChatItem_ db user chatDirection False Nothing (Just sharedMsgId) ciContent quoteRow Nothing timed False False hasLink createdAt Nothing Nothing Nothing Nothing (Just CIFLinked) (Just feedItemId) createdAt @@ -1309,6 +1307,7 @@ getChatContentTypes :: DB.Connection -> User -> ChatRef -> ExceptT StoreError IO getChatContentTypes db User {userId} (ChatRef cType chatId chatScope_) = case cType of CTDirect -> getTypes " contact_id = ? " () CTLocal -> getTypes " note_folder_id = ? " () + CTFeed -> getTypes " feed_id = ? " () CTGroup -> case chatScope_ of Nothing -> getTypes " group_id = ? AND group_scope_tag IS NULL AND group_scope_group_member_id IS NULL " () Just (GCSMemberSupport mId_) -> getTypes " group_id = ? AND group_scope_tag = ? AND group_scope_group_member_id IS NOT DISTINCT FROM ? " (GCSTMemberSupport_, mId_) @@ -1347,21 +1346,20 @@ getDirectChatLast_ db user ct contentFilter count search = do safeGetDirectItem :: DB.Connection -> User -> Contact -> UTCTime -> ChatItemId -> IO (CChatItem 'CTDirect) safeGetDirectItem db user ct currentTs itemId = runExceptT (getDirectCIWithReactions db user ct itemId) - >>= pure <$> safeToDirectItem currentTs itemId + >>= pure <$> safeToChatItem CIDirectSnd currentTs itemId -safeToDirectItem :: UTCTime -> ChatItemId -> Either StoreError (CChatItem 'CTDirect) -> CChatItem 'CTDirect -safeToDirectItem currentTs itemId = \case +safeToChatItem :: CIDirection c 'MDSnd -> UTCTime -> ChatItemId -> Either StoreError (CChatItem c) -> CChatItem c +safeToChatItem chatDir currentTs itemId = \case Right ci -> ci - Left e@(SEBadChatItem _ (Just itemTs)) -> badDirectItem itemTs e - Left e -> badDirectItem currentTs e + Left e@(SEBadChatItem _ (Just itemTs)) -> badChatItem itemTs e + Left e -> badChatItem currentTs e where - badDirectItem :: UTCTime -> StoreError -> CChatItem 'CTDirect - badDirectItem ts e = + badChatItem ts e = let errorText = T.pack $ show e in CChatItem SMDSnd ChatItem - { chatDir = CIDirectSnd, + { chatDir, meta = dummyMeta itemId ts errorText, content = CIInvalidJSON errorText, mentions = M.empty, @@ -1699,29 +1697,7 @@ getChatItemIDs db User {userId} cInfo contentFilter range count search = case cI safeGetGroupItem :: DB.Connection -> User -> GroupInfo -> UTCTime -> ChatItemId -> IO (CChatItem 'CTGroup) safeGetGroupItem db user g currentTs itemId = runExceptT (getGroupCIWithReactions db user g itemId) - >>= pure <$> safeToGroupItem currentTs itemId - -safeToGroupItem :: UTCTime -> ChatItemId -> Either StoreError (CChatItem 'CTGroup) -> CChatItem 'CTGroup -safeToGroupItem currentTs itemId = \case - Right ci -> ci - Left e@(SEBadChatItem _ (Just itemTs)) -> badGroupItem itemTs e - Left e -> badGroupItem currentTs e - where - badGroupItem :: UTCTime -> StoreError -> CChatItem 'CTGroup - badGroupItem ts e = - let errorText = T.pack $ show e - in CChatItem - SMDSnd - ChatItem - { chatDir = CIGroupSnd, - meta = dummyMeta itemId ts errorText, - content = CIInvalidJSON errorText, - mentions = M.empty, - formattedText = Nothing, - quotedItem = Nothing, - reactions = [], - file = Nothing - } + >>= pure <$> safeToChatItem CIGroupSnd currentTs itemId getGroupMemberChatItemLast :: DB.Connection -> User -> GroupId -> GroupMemberId -> ExceptT StoreError IO (CChatItem 'CTGroup) getGroupMemberChatItemLast db user@User {userId} groupId groupMemberId = do @@ -1939,29 +1915,7 @@ getLocalChatLast_ db user nf contentFilter count search = do safeGetLocalItem :: DB.Connection -> User -> NoteFolder -> UTCTime -> ChatItemId -> IO (CChatItem 'CTLocal) safeGetLocalItem db user NoteFolder {noteFolderId} currentTs itemId = runExceptT (getLocalChatItem db user noteFolderId itemId) - >>= pure <$> safeToLocalItem currentTs itemId - -safeToLocalItem :: UTCTime -> ChatItemId -> Either StoreError (CChatItem 'CTLocal) -> CChatItem 'CTLocal -safeToLocalItem currentTs itemId = \case - Right ci -> ci - Left e@(SEBadChatItem _ (Just itemTs)) -> badLocalItem itemTs e - Left e -> badLocalItem currentTs e - where - badLocalItem :: UTCTime -> StoreError -> CChatItem 'CTLocal - badLocalItem ts e = - let errorText = T.pack $ show e - in CChatItem - SMDSnd - ChatItem - { chatDir = CILocalSnd, - meta = dummyMeta itemId ts errorText, - content = CIInvalidJSON errorText, - mentions = M.empty, - formattedText = Nothing, - quotedItem = Nothing, - reactions = [], - file = Nothing - } + >>= pure <$> safeToChatItem CILocalSnd currentTs itemId getLocalChatAfter_ :: DB.Connection -> User -> NoteFolder -> Maybe MsgContentTag -> ChatItemId -> Int -> Text -> ExceptT StoreError IO (Chat 'CTLocal) getLocalChatAfter_ db user nf@NoteFolder {noteFolderId} contentFilter afterId count search = do @@ -2608,34 +2562,28 @@ getAllChatItems db cxt user@User {userId} pagination search_ = do |] (userId, CISRcvNew) --- The second query resolves a feed message delivery to the instance of the --- broadcast in the chat of the delivery's connection: a feed message is linked --- to the feed item, not to the instances. -getChatItemIdsByAgentMsgId :: DB.Connection -> Int64 -> AgentMsgId -> IO [ChatItemId] -getChatItemIdsByAgentMsgId db connId msgId = +getDirectChatItemIdsByAgentMsgId :: DB.Connection -> ContactId -> Int64 -> AgentMsgId -> IO [ChatItemId] +getDirectChatItemIdsByAgentMsgId db = getChatItemIdsByAgentMsgId_ db "contact_id" + +getGroupChatItemIdsByAgentMsgId :: DB.Connection -> GroupId -> Int64 -> AgentMsgId -> IO [ChatItemId] +getGroupChatItemIdsByAgentMsgId db = getChatItemIdsByAgentMsgId_ db "group_id" + +getChatItemIdsByAgentMsgId_ :: DB.Connection -> Query -> Int64 -> Int64 -> AgentMsgId -> IO [ChatItemId] +getChatItemIdsByAgentMsgId_ db chatIdColumn chatId connId msgId = map fromOnly <$> DB.query db - [sql| - SELECT chat_item_id - FROM chat_item_messages - WHERE message_id IN ( - SELECT message_id - FROM msg_deliveries - WHERE connection_id = ? AND agent_msg_id = ? - ) - UNION - SELECT i.chat_item_id - FROM msg_deliveries d - JOIN chat_item_messages cim ON cim.message_id = d.message_id - JOIN connections c ON c.connection_id = d.connection_id - LEFT JOIN group_members gm ON gm.group_member_id = c.group_member_id - JOIN chat_items i ON i.feed_item_id = cim.chat_item_id - AND ((c.contact_id IS NOT NULL AND i.contact_id = c.contact_id) - OR (gm.group_id IS NOT NULL AND i.group_id = gm.group_id)) - WHERE d.connection_id = ? AND d.agent_msg_id = ? - |] - (connId, msgId, connId, msgId) + ( "SELECT chat_item_id FROM chat_item_messages" + <> " WHERE message_id IN (SELECT message_id FROM msg_deliveries WHERE connection_id = ? AND agent_msg_id = ?)" + <> " UNION" + <> " SELECT i.chat_item_id FROM msg_deliveries d" + <> " JOIN chat_item_messages cim ON cim.message_id = d.message_id" + <> " JOIN chat_items i ON i.feed_item_id = cim.chat_item_id AND i." + <> chatIdColumn + <> " = ?" + <> " WHERE d.connection_id = ? AND d.agent_msg_id = ?" + ) + (connId, msgId, chatId, connId, msgId) updateDirectChatItemStatus :: forall d. MsgDirectionI d => DB.Connection -> User -> Contact -> ChatItemId -> CIStatus d -> ExceptT StoreError IO (ChatItem 'CTDirect d) updateDirectChatItemStatus db user@User {userId} ct@Contact {contactId} itemId itemStatus = do @@ -2778,7 +2726,7 @@ getDirectChatItemBySharedMsgId db user@User {userId} contactId sharedMsgId = do getDirectChatItemsByAgentMsgId :: DB.Connection -> User -> ContactId -> Int64 -> AgentMsgId -> IO [CChatItem 'CTDirect] getDirectChatItemsByAgentMsgId db user contactId connId msgId = do - itemIds <- getChatItemIdsByAgentMsgId db connId msgId + itemIds <- getDirectChatItemIdsByAgentMsgId db contactId connId msgId catMaybes <$> mapM (fmap eitherToMaybe . runExceptT . getDirectChatItem db user contactId) itemIds getDirectChatItemIdBySharedMsgId_ :: DB.Connection -> UserId -> Int64 -> SharedMsgId -> ExceptT StoreError IO Int64 @@ -3210,7 +3158,7 @@ getGroupMemberCIBySharedMsgId db user@User {userId} g@GroupInfo {groupId} member getGroupChatItemsByAgentMsgId :: DB.Connection -> User -> GroupId -> Int64 -> AgentMsgId -> IO [CChatItem 'CTGroup] getGroupChatItemsByAgentMsgId db user groupId connId msgId = do - itemIds <- getChatItemIdsByAgentMsgId db connId msgId + itemIds <- getGroupChatItemIdsByAgentMsgId db groupId connId msgId catMaybes <$> mapM (fmap eitherToMaybe . runExceptT . getGroupChatItem db user groupId) itemIds getGroupChatItem :: DB.Connection -> User -> Int64 -> ChatItemId -> ExceptT StoreError IO (CChatItem 'CTGroup) @@ -3350,7 +3298,6 @@ getLocalChatItem db User {userId} folderId itemId = ExceptT $ do |] (userId, folderId, itemId) --- a feed item is always sent; its reactions are aggregated from the instances toFeedChatItem :: UTCTime -> [CIReactionCount] -> ChatItemRow -> Either StoreError (CChatItem 'CTFeed) toFeedChatItem currentTs reactions ((itemId, itemTs, AMsgDirection msgDir, itemContentText, itemText, itemStatus, sentViaProxy, sharedMsgId) :. (itemDeleted, deletedTs, itemEdited, createdAt, updatedAt) :. forwardedFromRow :. (timedTTL, timedDeleteAt, itemLive, BI userMention, BI hasLink, msgSigned, itemFeed) :. (fileId_, fileName_, fileSize_, filePath, fileKey, fileNonce, fileStatus_, fileProtocol_, fileExpires)) = chatItem $ fromRight invalid $ dbParseACIContent itemContentText @@ -3416,7 +3363,6 @@ getFeedChatItem db user@User {userId} feedId itemId = ExceptT $ do |] (userId, feedId, itemId) --- reactions received on the instances of a broadcast, in every chat it was sent to getFeedCIReactions :: DB.Connection -> User -> SharedMsgId -> IO [CIReactionCount] getFeedCIReactions db User {userId} itemSharedMsgId = map toFeedReaction @@ -3439,29 +3385,7 @@ getFeedCIReactions db User {userId} itemSharedMsgId = safeGetFeedItem :: DB.Connection -> User -> Feed -> UTCTime -> ChatItemId -> IO (CChatItem 'CTFeed) safeGetFeedItem db user Feed {feedId} currentTs itemId = runExceptT (getFeedChatItem db user feedId itemId) - >>= pure <$> safeToFeedItem currentTs itemId - -safeToFeedItem :: UTCTime -> ChatItemId -> Either StoreError (CChatItem 'CTFeed) -> CChatItem 'CTFeed -safeToFeedItem currentTs itemId = \case - Right ci -> ci - Left e@(SEBadChatItem _ (Just itemTs)) -> badFeedItem itemTs e - Left e -> badFeedItem currentTs e - where - badFeedItem :: UTCTime -> StoreError -> CChatItem 'CTFeed - badFeedItem ts e = - let errorText = T.pack $ show e - in CChatItem - SMDSnd - ChatItem - { chatDir = CIFeedSnd, - meta = dummyMeta itemId ts errorText, - content = CIInvalidJSON errorText, - mentions = M.empty, - formattedText = Nothing, - quotedItem = Nothing, - reactions = [], - file = Nothing - } + >>= pure <$> safeToChatItem CIFeedSnd currentTs itemId getFeedChat :: DB.Connection -> User -> FeedId -> Maybe MsgContentTag -> ChatPagination -> Maybe Text -> ExceptT StoreError IO (Chat 'CTFeed, Maybe NavigationInfo) getFeedChat db user feedId contentFilter pagination search_ = do diff --git a/tests/Bots/BroadcastTests.hs b/tests/Bots/BroadcastTests.hs index 654fc00b6e..68043d73f2 100644 --- a/tests/Bots/BroadcastTests.hs +++ b/tests/Bots/BroadcastTests.hs @@ -80,7 +80,6 @@ testBroadcastMessages ps = do bob <## "I broadcast messages to all connected users from @alice." cath `connectVia` botLink alice #> "@broadcast_bot hello all!" - -- the bot replies as soon as the feed item is created; the broadcast is delivered by feed jobs alice <### [ WithTime "broadcast_bot> > hello all!", ConsoleString " Message is being delivered to all contacts", diff --git a/tests/ChatTests/Feed.hs b/tests/ChatTests/Feed.hs index d89f83f66f..8fadf097b0 100644 --- a/tests/ChatTests/Feed.hs +++ b/tests/ChatTests/Feed.hs @@ -15,7 +15,6 @@ chatFeedTests = do it "broadcast to contacts and customer groups in several buckets" testFeedBuckets it "edit and delete the broadcast in every chat" testFeedEditDelete --- one recipient per bucket, so each stream runs three buckets for two recipients feedTestCfg :: ChatConfig feedTestCfg = testCfg {feedBucketSize = 1} @@ -24,10 +23,8 @@ testFeedBuckets = testChatCfg5 feedTestCfg businessProfile aliceProfile bobProfile cathProfile danProfile $ \biz alice bob cath dan -> do createCCFeed biz - -- two contacts connectUsers biz alice connectUsers biz bob - -- two customer groups biz ##> "/ad" cLink <- getContactLink biz True biz ##> "/auto_accept on business" @@ -37,13 +34,11 @@ testFeedBuckets = biz `send` "/feed hello everyone" biz <# "% hello everyone" - -- the contacts stream and the customer groups stream deliver concurrently alice <# "biz> hello everyone" bob <# "biz> hello everyone" cath <# "#biz biz_1> hello everyone" dan <# "#biz biz_1> hello everyone" - -- the broadcast is one item in the feed and the last item of every recipient chat chatItems biz "%1" 10 `shouldReturn` [(1, "hello everyone")] chatItems biz "@2" 1 `shouldReturn` [(1, "hello everyone")] chatItems biz "@3" 1 `shouldReturn` [(1, "hello everyone")] @@ -69,7 +64,6 @@ testFeedEditDelete = bob <# "biz> hello everyone" cath <# "#biz biz_1> hello everyone" - -- the edit applies to the instance of every chat biz ##> "! % (hello everyone) hello again" biz <# "% [edited] hello again" alice <# "biz> [edited] hello again" @@ -80,9 +74,7 @@ testFeedEditDelete = chatItems biz "@3" 1 `shouldReturn` [(1, "hello again")] chatItems biz "#1" 1 `shouldReturn` [(1, "hello again")] - -- the feed item is removed when every instance is marked deleted biz ##> "\\ % hello again" - -- the response of the command and the event of the last job biz <### [ConsoleString "message being deleted", ConsoleString "message deleted"] alice <# "biz> [marked deleted] hello again" bob <# "biz> [marked deleted] hello again" diff --git a/tests/ProtocolTests.hs b/tests/ProtocolTests.hs index 78393e093c..5504eb5d0a 100644 --- a/tests/ProtocolTests.hs +++ b/tests/ProtocolTests.hs @@ -247,9 +247,15 @@ decodeChatMessageTest = describe "Chat message encoding/decoding" $ do it "x.msg.new forward with file" $ "{\"v\":\"9\",\"msgId\":\"AQIDBA==\",\"event\":\"x.msg.new\",\"params\":{\"content\":{\"text\":\"hello\",\"type\":\"text\"},\"file\":{\"fileSize\":12345,\"fileName\":\"photo.jpg\"},\"forward\":true}}" ##==## ChatMessage chatInitialVRange (Just $ SharedMsgId "\1\2\3\4") (XMsgNew $ (mcForward Nothing (MCText "hello")) {file = Just FileInvitation {fileName = "photo.jpg", fileSize = 12345, fileDigest = Nothing, fileConnReq = Nothing, fileInline = Nothing, fileDescr = Nothing}}) + it "x.msg.new broadcast" $ + "{\"v\":\"9\",\"msgId\":\"AQIDBA==\",\"event\":\"x.msg.new\",\"params\":{\"content\":{\"text\":\"hello\",\"type\":\"text\"},\"feed\":true}}" + ##==## ChatMessage chatInitialVRange (Just $ SharedMsgId "\1\2\3\4") (XMsgNew $ (mcSimple (MCText "hello")) {feed = Just True}) it "x.msg.update" $ "{\"v\":\"9\",\"event\":\"x.msg.update\",\"params\":{\"msgId\":\"AQIDBA==\", \"content\":{\"text\":\"hello\",\"type\":\"text\"}}}" #==# XMsgUpdate (SharedMsgId "\1\2\3\4") (MCText "hello") [] Nothing Nothing Nothing Nothing Nothing + it "x.msg.update broadcast" $ + "{\"v\":\"9\",\"event\":\"x.msg.update\",\"params\":{\"msgId\":\"AQIDBA==\", \"content\":{\"text\":\"hello\",\"type\":\"text\"},\"feed\":true}}" + #==# XMsgUpdate (SharedMsgId "\1\2\3\4") (MCText "hello") [] Nothing Nothing Nothing Nothing (Just True) it "x.msg.del" $ "{\"v\":\"9\",\"event\":\"x.msg.del\",\"params\":{\"msgId\":\"AQIDBA==\"}}" #==# XMsgDel (SharedMsgId "\1\2\3\4") Nothing Nothing False