fixes

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

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.

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
 

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 <text>": 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))),

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

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

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

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

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

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",

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"

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