From 953683c289248882a234d51f7f874992dd3182ba Mon Sep 17 00:00:00 2001 From: "Evgeny @ SimpleX Chat" <259188159+evgeny-simplex@users.noreply.github.com> Date: Fri, 11 Sep 2026 17:46:10 +0000 Subject: [PATCH] split jobs --- bots/api/TYPES.md | 8 + .../types/typescript/src/types.ts | 14 + .../src/simplex_chat/types/_types.py | 12 +- plans/2026-09-04-feed-broadcast-chat.md | 5 + plans/2026-09-11-feed-jobs-split.md | 87 ++++-- src/Simplex/Chat.hs | 2 + src/Simplex/Chat/Controller.hs | 3 +- src/Simplex/Chat/Delivery.hs | 62 +++-- src/Simplex/Chat/Library/Commands.hs | 97 ++++--- src/Simplex/Chat/Library/Internal.hs | 30 +- src/Simplex/Chat/Library/Subscriber.hs | 263 ++++++++++-------- src/Simplex/Chat/Protocol.hs | 12 +- src/Simplex/Chat/Store/Delivery.hs | 169 ++--------- src/Simplex/Chat/Store/Direct.hs | 1 + src/Simplex/Chat/Store/Feeds.hs | 154 +++++++++- .../Postgres/Migrations/M20260905_feeds.hs | 36 +-- .../SQLite/Migrations/M20260905_feeds.hs | 48 ++-- .../Store/SQLite/Migrations/chat_schema.sql | 29 +- src/Simplex/Chat/Store/Shared.hs | 2 + src/Simplex/Chat/View.hs | 4 + tests/ChatTests/Feed.hs | 117 ++++++++ tests/ChatTests/Groups.hs | 2 +- tests/ProtocolTests.hs | 5 +- 23 files changed, 720 insertions(+), 442 deletions(-) diff --git a/bots/api/TYPES.md b/bots/api/TYPES.md index be6eb2a702..efa490524d 100644 --- a/bots/api/TYPES.md +++ b/bots/api/TYPES.md @@ -4282,6 +4282,14 @@ DeliveryJobNotFound: - type: "deliveryJobNotFound" - jobId: int64 +InvalidFeedJob: +- type: "invalidFeedJob" +- feedJobId: int64 + +FeedJobNotFound: +- type: "feedJobNotFound" +- feedJobId: int64 + WorkItemError: - type: "workItemError" - errContext: string diff --git a/packages/simplex-chat-client/types/typescript/src/types.ts b/packages/simplex-chat-client/types/typescript/src/types.ts index 88eabac153..674bf5c62e 100644 --- a/packages/simplex-chat-client/types/typescript/src/types.ts +++ b/packages/simplex-chat-client/types/typescript/src/types.ts @@ -4438,6 +4438,8 @@ export type StoreError = | StoreError.DeliveryTaskNotFound | StoreError.InvalidDeliveryJob | StoreError.DeliveryJobNotFound + | StoreError.InvalidFeedJob + | StoreError.FeedJobNotFound | StoreError.WorkItemError export namespace StoreError { @@ -4533,6 +4535,8 @@ export namespace StoreError { | "deliveryTaskNotFound" | "invalidDeliveryJob" | "deliveryJobNotFound" + | "invalidFeedJob" + | "feedJobNotFound" | "workItemError" interface Interface { @@ -4979,6 +4983,16 @@ export namespace StoreError { jobId: number // int64 } + export interface InvalidFeedJob extends Interface { + type: "invalidFeedJob" + feedJobId: number // int64 + } + + export interface FeedJobNotFound extends Interface { + type: "feedJobNotFound" + feedJobId: number // int64 + } + export interface WorkItemError extends Interface { type: "workItemError" errContext: string diff --git a/packages/simplex-chat-python/src/simplex_chat/types/_types.py b/packages/simplex-chat-python/src/simplex_chat/types/_types.py index 0818c4d51c..3025464c22 100644 --- a/packages/simplex-chat-python/src/simplex_chat/types/_types.py +++ b/packages/simplex-chat-python/src/simplex_chat/types/_types.py @@ -3390,6 +3390,14 @@ class StoreError_deliveryJobNotFound(TypedDict): type: Literal["deliveryJobNotFound"] jobId: int # int64 +class StoreError_invalidFeedJob(TypedDict): + type: Literal["invalidFeedJob"] + feedJobId: int # int64 + +class StoreError_feedJobNotFound(TypedDict): + type: Literal["feedJobNotFound"] + feedJobId: int # int64 + class StoreError_workItemError(TypedDict): type: Literal["workItemError"] errContext: str @@ -3486,10 +3494,12 @@ StoreError = ( | StoreError_deliveryTaskNotFound | StoreError_invalidDeliveryJob | StoreError_deliveryJobNotFound + | StoreError_invalidFeedJob + | StoreError_feedJobNotFound | StoreError_workItemError ) -StoreError_Tag = Literal["duplicateName", "userNotFound", "relayUserNotFound", "userNotFoundByName", "userNotFoundByContactId", "userNotFoundByGroupId", "userNotFoundByFileId", "userNotFoundByContactRequestId", "contactNotFound", "contactNotFoundByName", "contactNotFoundByMemberId", "contactNotReady", "duplicateContactLink", "userContactLinkNotFound", "contactRequestNotFound", "contactRequestNotFoundByName", "invalidContactRequestEntity", "invalidBusinessChatContactRequest", "groupNotFound", "groupNotFoundByName", "groupMemberNameNotFound", "groupMemberNotFound", "groupMemberNotFoundByIndex", "memberRelationsVectorNotFound", "groupHostMemberNotFound", "groupMemberNotFoundByMemberId", "memberContactGroupMemberNotFound", "invalidMemberRelationUpdate", "groupWithoutUser", "duplicateGroupMember", "duplicateMemberId", "groupAlreadyJoined", "groupInvitationNotFound", "noteFolderAlreadyExists", "noteFolderNotFound", "userNoteFolderNotFound", "feedAlreadyExists", "feedNotFound", "userFeedNotFound", "sndFileNotFound", "sndFileInvalid", "rcvFileNotFound", "rcvFileDescrNotFound", "fileNotFound", "rcvFileInvalid", "rcvFileInvalidDescrPart", "localFileNoTransfer", "sharedMsgIdNotFoundByFileId", "fileIdNotFoundBySharedMsgId", "sndFileNotFoundXFTP", "rcvFileNotFoundXFTP", "connectionNotFound", "connectionNotFoundById", "connectionNotFoundByMemberId", "pendingConnectionNotFound", "uniqueID", "largeMsg", "internalError", "dBException", "dBBusyError", "badChatItem", "chatItemNotFound", "chatItemNotFoundByText", "chatItemSharedMsgIdNotFound", "chatItemNotFoundByFileId", "chatItemNotFoundByContactId", "chatItemNotFoundByGroupId", "profileNotFound", "duplicateGroupLink", "groupLinkNotFound", "hostMemberIdNotFound", "contactNotFoundByFileId", "noGroupSndStatus", "duplicateGroupMessage", "remoteHostNotFound", "remoteHostUnknown", "remoteHostDuplicateCA", "remoteCtrlNotFound", "remoteCtrlDuplicateCA", "prohibitedDeleteUser", "operatorNotFound", "usageConditionsNotFound", "userChatRelayNotFound", "groupRelayNotFound", "groupRelayNotFoundByMemberId", "invalidQuote", "invalidMention", "invalidDeliveryTask", "deliveryTaskNotFound", "invalidDeliveryJob", "deliveryJobNotFound", "workItemError"] +StoreError_Tag = Literal["duplicateName", "userNotFound", "relayUserNotFound", "userNotFoundByName", "userNotFoundByContactId", "userNotFoundByGroupId", "userNotFoundByFileId", "userNotFoundByContactRequestId", "contactNotFound", "contactNotFoundByName", "contactNotFoundByMemberId", "contactNotReady", "duplicateContactLink", "userContactLinkNotFound", "contactRequestNotFound", "contactRequestNotFoundByName", "invalidContactRequestEntity", "invalidBusinessChatContactRequest", "groupNotFound", "groupNotFoundByName", "groupMemberNameNotFound", "groupMemberNotFound", "groupMemberNotFoundByIndex", "memberRelationsVectorNotFound", "groupHostMemberNotFound", "groupMemberNotFoundByMemberId", "memberContactGroupMemberNotFound", "invalidMemberRelationUpdate", "groupWithoutUser", "duplicateGroupMember", "duplicateMemberId", "groupAlreadyJoined", "groupInvitationNotFound", "noteFolderAlreadyExists", "noteFolderNotFound", "userNoteFolderNotFound", "feedAlreadyExists", "feedNotFound", "userFeedNotFound", "sndFileNotFound", "sndFileInvalid", "rcvFileNotFound", "rcvFileDescrNotFound", "fileNotFound", "rcvFileInvalid", "rcvFileInvalidDescrPart", "localFileNoTransfer", "sharedMsgIdNotFoundByFileId", "fileIdNotFoundBySharedMsgId", "sndFileNotFoundXFTP", "rcvFileNotFoundXFTP", "connectionNotFound", "connectionNotFoundById", "connectionNotFoundByMemberId", "pendingConnectionNotFound", "uniqueID", "largeMsg", "internalError", "dBException", "dBBusyError", "badChatItem", "chatItemNotFound", "chatItemNotFoundByText", "chatItemSharedMsgIdNotFound", "chatItemNotFoundByFileId", "chatItemNotFoundByContactId", "chatItemNotFoundByGroupId", "profileNotFound", "duplicateGroupLink", "groupLinkNotFound", "hostMemberIdNotFound", "contactNotFoundByFileId", "noGroupSndStatus", "duplicateGroupMessage", "remoteHostNotFound", "remoteHostUnknown", "remoteHostDuplicateCA", "remoteCtrlNotFound", "remoteCtrlDuplicateCA", "prohibitedDeleteUser", "operatorNotFound", "usageConditionsNotFound", "userChatRelayNotFound", "groupRelayNotFound", "groupRelayNotFoundByMemberId", "invalidQuote", "invalidMention", "invalidDeliveryTask", "deliveryTaskNotFound", "invalidDeliveryJob", "deliveryJobNotFound", "invalidFeedJob", "feedJobNotFound", "workItemError"] class SubscriptionStatus_active(TypedDict): type: Literal["active"] diff --git a/plans/2026-09-04-feed-broadcast-chat.md b/plans/2026-09-04-feed-broadcast-chat.md index 772fd0a622..99040811eb 100644 --- a/plans/2026-09-04-feed-broadcast-chat.md +++ b/plans/2026-09-04-feed-broadcast-chat.md @@ -1,5 +1,10 @@ # Feed: a broadcast chat +Superseded by `plans/2026-09-11-feed-jobs-split.md` wherever the two differ. +Feed jobs are held in their own `feed_jobs` table with their own worker key, +so every reference below to `delivery_jobs`, `DeliveryJobKey`, `DeliveryJobWork` +and `job_scope_spec_tag` for feeds describes the earlier design. + Status: the core is implemented and compiles; the schema tests pass and the schema dump is regenerated. Deviations from the sections below, made while implementing: diff --git a/plans/2026-09-11-feed-jobs-split.md b/plans/2026-09-11-feed-jobs-split.md index 230d504eb7..5e980565a7 100644 --- a/plans/2026-09-11-feed-jobs-split.md +++ b/plans/2026-09-11-feed-jobs-split.md @@ -108,7 +108,7 @@ Feed job statements move to `src/Simplex/Chat/Store/Feeds.hs`: - `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`. +- `notification` derives from `feedActionCreates action`; `feedActionEventTag` is removed. - `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. @@ -119,28 +119,71 @@ Feed job statements move to `src/Simplex/Chat/Store/Feeds.hs`: ### 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. +1. **An inner exception must not fail a job.** The `FJANew` bucket creates every instance in one + `withStore'`, so a foreign key violation from a contact, group or member deleted between the bucket's + read and its insert loses the rest of the broadcast. Write instances with `withStoreBatch'` per + recipient and drop the failing recipient. Same for `createGroupSndStatus`. +2. **An errored feed job resumes after restart.** `resumeFeedJobsOnStart` resets + `failed = 0 AND job_status = DJSError` to `DJSPending` before the workers are started, and the job + continues from its stored cursor. Within a session an errored job is not retried, so the worker does + not spin. `failed = 1` stays terminal: it marks a row that failed to decode. +3. **A job error raises a critical error.** `CRITICAL {offerRestart = True}`, as at `Subscriber.hs:153` + and `Commands.hs:5316`, so the app offers a restart. +4. **The feed item's status is reported on every change.** `CISSndError` on a job error and + `CISSndSent SSPPartial` on the first bucket each emit `CEvtChatItemsStatusesUpdated`. +5. **`finishFeedEvent` runs inside the feed lock**, before `completeFeedJob` writes `DJSComplete`. +6. **Group instance statuses are recorded per member.** A member's send failure writes `GSSError` against + that member instead of `CISSndError` on the whole instance. Members without a ready connection are + still skipped, and no pending messages are stored. +7. **`APISendFeedMessage` creates the item, the message, the message link and the jobs in one transaction.** +8. **`CIDMInternal` is accepted on a marked feed item**, as in direct and group chats. +9. **`feed :: Maybe Bool` is added to `XMsgDel` and `XMsgFileDescr`**, as it was to `XMsgUpdate`. The + guards in `messageDelete`, `groupMessageDelete` and `skipDroppedFeedFile` become + `dropsFeed chatSettings feed_`. +10. **`CTFeed` is accepted by `SendFile` and `SendImage`**, as `CTLocal` already is. +11. **`setUserChatsRead` clears `feeds.unread_chat`.** +12. **`SEUserFeedNotFound` and `SEFeedNotFound` get `viewStoreError` cases.** +13. **`unzipMaybe3` is renamed** — it unzips a pair. +14. **`chat_query_plans.txt` is regenerated** from a full test run. +15. **`plans/2026-09-04-feed-broadcast-chat.md` is updated or dropped** — it contradicts the code. +16. **Tests** 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. +17. **Feed job cleanup** is wired into the cleanup manager as `deleteDoneFeedJobs`. +18. **The file description path is one transaction** — `createFeedFileDescrJobs` creates every part + message, its link to the feed item and the jobs together. +19. **`deleteFeedInstances` no longer deletes `chat_item_messages`** — an instance is created without a + message link, and the row would cascade with the item. -## 6. Order of work +### Not changed + +- `feeds.favorite` stays, mirroring `NoteFolder`: `note_folders.favorite` has no writer either, so the + unwritten column is pre-existing rather than introduced here. +- `getChatRefViaItemId` stays as it is. `APIGetChatItems` across all chats is used only from the CLI, and + its pagination is unused. +- Signing in customer groups: `signMessages` defaults to `FEOff` and feed groups are filtered to + `business_chat = BCCustomer`, so an unsigned broadcast never reaches a signing channel. +- Retry and the critical error for group delivery jobs: the same gap exists there, outside this change. +- `createContactPQSndItem` on the feed path: `deliverMessagesB` already updates the stored PQ state, so + only the chat item announcing the change is missing. +- A dropped broadcast still stores its body in `messages` for 30 days and still returns a delivery + receipt — documented, not changed. + +## 6. Done + +Sections 1 to 4 are implemented, and items 1 to 13, 15, and 17 to 19 of section 5. + +Item 16: tests cover `dropFeed`, a file broadcast and a per-chat detach. A restart in the middle of a +job, incognito exclusion and a member send error remain uncovered — none of the three can be made +deterministic without an induced failure. + +Item 14 needs a full test run with write access to `chat_query_plans.txt`. + +The Postgres schema dump is stale: `src/Simplex/Chat/Store/Postgres/Migrations/chat_schema.sql` still +holds the feed columns on `delivery_jobs` and no `feed_jobs`. Only the Postgres client binaries are +installed here, so `postgresSchemaDumpTest` cannot be run. + +## 7. Order of work 1. Amend the migrations. 2. Split the types. diff --git a/src/Simplex/Chat.hs b/src/Simplex/Chat.hs index f53522096c..a225471eaf 100644 --- a/src/Simplex/Chat.hs +++ b/src/Simplex/Chat.hs @@ -188,6 +188,7 @@ newChatController chatStoreChanged <- newTVarIO False deliveryTaskWorkers <- TM.emptyIO deliveryJobWorkers <- TM.emptyIO + feedJobWorkers <- TM.emptyIO relayRequestWorkers <- TM.emptyIO relayGroupLinkChecksAsync <- newTVarIO Nothing webPreviewState <- forM webPreviewConfig $ \_ -> newWebPreviewState @@ -234,6 +235,7 @@ newChatController filesFolder, deliveryTaskWorkers, deliveryJobWorkers, + feedJobWorkers, relayRequestWorkers, relayGroupLinkChecksAsync, webPreviewState, diff --git a/src/Simplex/Chat/Controller.hs b/src/Simplex/Chat/Controller.hs index 77293b9f4a..1d8a8f9cbf 100644 --- a/src/Simplex/Chat/Controller.hs +++ b/src/Simplex/Chat/Controller.hs @@ -307,7 +307,8 @@ data ChatController = ChatController config :: ChatConfig, filesFolder :: TVar (Maybe FilePath), -- path to files folder for mobile apps, deliveryTaskWorkers :: TMap DeliveryWorkerKey Worker, - deliveryJobWorkers :: TMap DeliveryJobKey Worker, + deliveryJobWorkers :: TMap DeliveryWorkerKey Worker, + feedJobWorkers :: TMap FeedJobKey Worker, relayRequestWorkers :: TMap Int Worker, -- single global worker with key 1 is used to fit into existing worker management framework relayGroupLinkChecksAsync :: TVar (Maybe (Async ())), webPreviewState :: Maybe WebPreviewState, diff --git a/src/Simplex/Chat/Delivery.hs b/src/Simplex/Chat/Delivery.hs index 278a13be78..5dc9bd9199 100644 --- a/src/Simplex/Chat/Delivery.hs +++ b/src/Simplex/Chat/Delivery.hs @@ -1,11 +1,8 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE DuplicateRecordFields #-} -{-# LANGUAGE GADTs #-} -{-# LANGUAGE KindSignatures #-} {-# LANGUAGE LambdaCase #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE StandaloneDeriving #-} module Simplex.Chat.Delivery where @@ -14,14 +11,17 @@ import Data.Int (Int64) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as L import Data.Maybe (fromMaybe) +import Data.Text (Text) +import qualified Data.Text as T import Data.Time.Clock (UTCTime) -import Simplex.Chat.Messages (ChatItemId, ChatType (..), GroupChatScopeInfo (..), MessageId, ShowGroupAsSender) +import Simplex.Chat.Messages (ChatItemId, GroupChatScopeInfo (..), MessageId, ShowGroupAsSender) import Simplex.Chat.Options.DB (FromField (..), ToField (..)) import Simplex.Chat.Protocol import Simplex.Chat.Types import Simplex.Chat.Types.Shared import Simplex.Messaging.Agent.Store.DB (fromTextField_) import Simplex.Messaging.Encoding.String +import Text.Read (readMaybe) type DeliveryWorkerKey = (GroupId, DeliveryWorkerScope) @@ -46,10 +46,7 @@ instance TextEncoding DeliveryWorkerScope where DWSMemberSupport -> "member_support" -- DWSMemberProfileUpdate -> "member_profile_update" -data DeliveryJobKey - = DJKGroup GroupId DeliveryWorkerScope - | DJKFeed FeedId FeedWorkerScope - deriving (Eq, Ord, Show) +type FeedJobKey = (FeedId, FeedWorkerScope) data FeedWorkerScope = FWSContacts | FWSGroups deriving (Eq, Ord, Show) @@ -187,22 +184,38 @@ instance TextEncoding DeliveryTaskStatus where DTSProcessed -> "processed" DTSError -> "error" -data DeliveryJob (c :: ChatType) = DeliveryJob +-- NULL and empty string mean []; any other value must parse as a comma-separated Int64 list. +parseIds :: Maybe Text -> Maybe [Int64] +parseIds = \case + Nothing -> Just [] + Just t + | T.null t -> Just [] + | otherwise -> traverse (readMaybe . T.unpack) (T.splitOn "," t) + +idsColumn :: [Int64] -> Maybe Text +idsColumn ids + | null ids = Nothing + | otherwise = Just $ T.intercalate "," $ map (T.pack . show) ids + +data MessageDeliveryJob = MessageDeliveryJob { jobId :: Int64, - cursorId_ :: Maybe Int64, - jobWork :: DeliveryJobWork c + jobScope :: DeliveryJobScope, + senderGMIds :: [GroupMemberId], + body :: ByteString, + cursorGMId_ :: Maybe GroupMemberId } + deriving (Show) -deriving instance Show (DeliveryJob c) +deliveryJobId :: MessageDeliveryJob -> Int64 +deliveryJobId = jobId -deliveryJobId :: DeliveryJob c -> Int64 -deliveryJobId DeliveryJob {jobId} = jobId - -data DeliveryJobWork (c :: ChatType) where - DJWGroup :: {jobScope :: DeliveryJobScope, senderGMIds :: [GroupMemberId], body :: ByteString} -> DeliveryJobWork 'CTGroup - DJWFeed :: {feedItemId :: ChatItemId, feedAction :: FeedJobAction} -> DeliveryJobWork 'CTFeed - -deriving instance Show (DeliveryJobWork c) +data FeedJob = FeedJob + { feedJobId :: Int64, + feedItemId :: ChatItemId, + feedAction :: FeedJobAction, + cursorId_ :: Maybe Int64 + } + deriving (Show) data FeedJobAction = FJANew MessageId @@ -240,15 +253,6 @@ feedActionMsgIds = \case FJADeleteInternal -> [] FJADeleteMark -> [] -feedActionEventTag :: FeedJobAction -> CMEventTag 'Json -feedActionEventTag = \case - FJANew _ -> XMsgNew_ - FJAFileDescr _ -> XMsgFileDescr_ - FJAUpdate _ -> XMsgUpdate_ - FJADeleteBroadcast _ -> XMsgDel_ - FJADeleteInternal -> XMsgDel_ - FJADeleteMark -> XMsgDel_ - feedActionCreates :: FeedJobAction -> Bool feedActionCreates = \case FJANew _ -> True diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index e651237cf8..eee044fbb5 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -60,7 +60,7 @@ import Simplex.Chat.Badges (BadgeCredential (..), LocalBadge (..), badgeServerCr import Simplex.Chat.Names (SimplexDomainProof (..), SimplexDomainClaim (..), claimDomain, mkDomainClaim) import Simplex.Chat.Call import Simplex.Chat.Controller -import Simplex.Chat.Delivery (DeliveryJobKey (..), DeliveryJobScope (..), DeliveryJobSpec (..), DeliveryWorkerScope (..), FeedJobAction (..)) +import Simplex.Chat.Delivery (DeliveryJobScope (..), DeliveryJobSpec (..), DeliveryWorkerScope (..), FeedJobAction (..)) import Simplex.Chat.Files import Simplex.Chat.Markdown import Simplex.Chat.Messages @@ -263,7 +263,7 @@ startChatController mainApp enableSndFiles serviceRequests = do Left e -> liftIO $ putStrLn $ "Error starting XFTP workers: " <> show e Right _ -> pure () startDeliveryWorkers = - runExceptT (startDeliveryTaskWorkers >> startDeliveryJobWorkers) >>= \case + runExceptT (startDeliveryTaskWorkers >> startDeliveryJobWorkers >> startFeedJobWorkers) >>= \case Left e -> liftIO $ putStrLn $ "Error starting delivery workers: " <> show e Right _ -> pure () startRelayRequestWorker_ = @@ -768,26 +768,26 @@ processChatCommand cxt nm = \case feed <- withFastStore $ \db -> getFeed db user feedId sharedMsgId <- getSharedMsgId createdAt <- liftIO getCurrentTime - (fInv_, ciFile_) <- unzipMaybe3 <$> setupSndFileTransfer user feed + (fInv_, ciFile_) <- unzipMaybe <$> setupSndFileTransfer user feed + mkMsg <- feedMessageMaker let hasLink = msgContentHasLink mc $ snd $ msgContentTexts mc - feedItemId <- withFastStore' $ \db -> do - void $ updateChatTsStats db cxt user (CDFeedSnd feed) createdAt Nothing - itemId <- createNewChatItemNoMsg db user (CDFeedSnd feed) False (CISndMsgContent mc) (Just sharedMsgId) hasLink Nothing createdAt createdAt - forM_ ciFile_ $ \CIFile {fileId} -> updateFileTransferChatItemId db fileId itemId createdAt - pure itemId - let mc' = (mcSimple mc) {file = fInv_, feed = Just True} - msg <- createFeedMessage feed (Just sharedMsgId) feedItemId $ XMsgNew mc' - withFastStore' $ \db -> createFeedJobs db feedId feedItemId (FJANew $ msgId' msg) + mc' = (mcSimple mc) {file = fInv_, feed = Just True} + ci <- withFastStore $ \db -> do + liftIO $ void $ updateChatTsStats db cxt user (CDFeedSnd feed) createdAt Nothing + feedItemId <- liftIO $ createNewChatItemNoMsg db user (CDFeedSnd feed) False (CISndMsgContent mc) (Just sharedMsgId) hasLink Nothing createdAt createdAt + liftIO $ forM_ ciFile_ $ \CIFile {fileId} -> updateFileTransferChatItemId db fileId feedItemId createdAt + msg <- mkMsg db feed (Just sharedMsgId) feedItemId (XMsgNew mc') createdAt + liftIO $ createFeedJobs db feedId feedItemId (FJANew $ msgId' msg) + getFeedChatItem db user feedId feedItemId startFeedWorkers feedId - ci <- withFastStore $ \db -> getFeedChatItem db user feedId feedItemId pure $ CRNewChatItems user [aFeedItem feed ci] where setupSndFileTransfer user@User {profile = LocalProfile {localBadge}} feed = forM file_ $ \file -> do fileSize <- checkSndFile localBadge file (fInv, ciFile, _) <- xftpSndFileTransfer_ user file fileSize 1 (Just $ CGFeed feed) pure (fInv, ciFile) - unzipMaybe3 :: Maybe (a, b) -> (Maybe a, Maybe b) - unzipMaybe3 = maybe (Nothing, Nothing) (\(a, b) -> (Just a, Just b)) + unzipMaybe :: Maybe (a, b) -> (Maybe a, Maybe b) + unzipMaybe = maybe (Nothing, Nothing) (\(a, b) -> (Just a, Just b)) APIReportMessage gId reportedItemId reportReason reportText -> withUser $ \user -> withGroupLock "reportMessage" gId $ do gInfo <- withFastStore $ \db -> getGroupInfo db cxt user gId @@ -885,12 +885,14 @@ processChatCommand cxt nm = \case | mc == oldMC -> pure $ CRChatItemNotChanged user (aFeedItem feed cci) | otherwise -> do let event = XMsgUpdate itemSharedMId mc M.empty Nothing Nothing Nothing Nothing (Just True) - msg <- createFeedMessage feed Nothing itemId event - ci' <- withFastStore' $ \db -> do - currentTs <- getCurrentTime - addInitialAndNewCIVersions db itemId (chatItemTs' ci, oldMC) (currentTs, mc) - createFeedJobs db chatId itemId (FJAUpdate $ msgId' msg) - updateFeedChatItem' db user chatId ci (CISndMsgContent mc) (msgContentHasLink mc $ snd $ msgContentTexts mc) + mkMsg <- feedMessageMaker + currentTs <- liftIO getCurrentTime + ci' <- withFastStore $ \db -> do + msg <- mkMsg db feed Nothing itemId event currentTs + liftIO $ do + addInitialAndNewCIVersions db itemId (chatItemTs' ci, oldMC) (currentTs, mc) + createFeedJobs db chatId itemId (FJAUpdate $ msgId' msg) + updateFeedChatItem' db user chatId ci (CISndMsgContent mc) (msgContentHasLink mc $ snd $ msgContentTexts mc) startFeedWorkers chatId pure $ CRChatItemUpdated user (AChatItem SCTFeed SMDSnd (FeedChat feed) ci') _ -> throwChatError CEInvalidChatItemUpdate @@ -910,7 +912,7 @@ processChatCommand cxt nm = \case assertDeletable items assertDirectAllowed user MDSnd ct XMsgDel_ let msgIds = itemsMsgIds items - events = map (\msgId -> XMsgDel msgId Nothing Nothing False) msgIds + events = map (\msgId -> XMsgDel msgId Nothing Nothing False Nothing) msgIds forM_ (L.nonEmpty events) $ \events' -> sendDirectContactMessages user ct events' if featureAllowed SCFFullDelete forUser ct @@ -948,22 +950,27 @@ processChatCommand cxt nm = \case CTFeed -> withFeedLock "deleteChatItem" chatId $ do (feed, items) <- getCommandFeedChatItems user chatId itemIds when (null items) $ throwChatError CEInvalidChatItemDelete + mkMsg <- feedMessageMaker deletions <- forM items $ \cci -> case cci of CChatItem SMDSnd ci@ChatItem {meta = CIMeta {itemId, itemSharedMsgId, itemTs, itemDeleted}} -> do deletedTs <- liftIO getCurrentTime - action <- case (mode, itemDeleted) of + delEvent <- case (mode, itemDeleted) of (CIDMHistory, _) -> throwChatError CEInvalidChatItemDelete + (CIDMInternal, _) -> pure $ Right FJADeleteInternal (_, 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) + Just sharedMsgId | diffUTCTime deletedTs itemTs < deleteMsgInterval -> + pure $ Left $ XMsgDel sharedMsgId Nothing Nothing False (Just True) _ -> throwChatError CEInvalidChatItemDelete - (CIDMInternalMark, Nothing) -> pure FJADeleteMark - _ -> pure FJADeleteInternal - ci' <- withFastStore' $ \db -> do - createFeedJobs db chatId itemId action - markFeedChatItemDeleted db user chatId ci (feedItemDeleted action deletedTs) deletedTs + (CIDMInternalMark, Nothing) -> pure $ Right FJADeleteMark + _ -> pure $ Right FJADeleteInternal + ci' <- withFastStore $ \db -> do + action <- case delEvent of + Left event -> FJADeleteBroadcast . msgId' <$> mkMsg db feed Nothing itemId event deletedTs + Right a -> pure a + liftIO $ do + createFeedJobs db chatId itemId action + 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 @@ -995,7 +1002,7 @@ processChatCommand cxt nm = \case delEvent <$> itemSharedMsgId where delEvent msgId = - let evt = XMsgDel msgId Nothing (toMsgScope gInfo <$> chatScopeInfo) onlyHistory + let evt = XMsgDel msgId Nothing (toMsgScope gInfo <$> chatScopeInfo) onlyHistory Nothing in (groupMsgSigning (onlyHistory || itemSigned) gInfo evt, evt) itemSigned = case msgVerified of Just (MVSigned _) -> True; _ -> False APIDeleteMemberChatItem gId itemIds -> withUser $ \user -> withGroupLock "deleteChatItem" gId $ do @@ -3276,7 +3283,7 @@ processChatCommand cxt nm = \case deleteGroupDeliveryTasks db gInfo deleteGroupDeliveryJobs db gInfo createMsgDeliveryJob db gInfo (DJSGroup {jobSpec = DJRelayRemoved}) [] body - lift . void $ getDeliveryJobWorker True (DJKGroup groupId DWSGroup) + lift . void $ getDeliveryJobWorker True (groupId, DWSGroup) pure msg leaveGroupSendMsg user gInfo = do (members, recipients) <- getRecipients user gInfo @@ -3538,16 +3545,19 @@ processChatCommand cxt nm = \case chatRef <- getChatRef user chatName case chatRef of ChatRef CTLocal folderId _ -> processChatCommand cxt nm $ APICreateChatItems folderId [composedMessage (Just f) (MCFile "")] + ChatRef CTFeed feedId _ -> processChatCommand cxt nm $ APISendFeedMessage feedId (composedMessage (Just f) (MCFile "")) _ -> withSendRef user chatRef $ \sendRef -> processChatCommand cxt nm $ APISendMessages sendRef False Nothing False [composedMessage (Just f) (MCFile "")] SendImage chatName f@(CryptoFile fPath _) -> withUser $ \user -> do chatRef <- getChatRef user chatName - withSendRef user chatRef $ \sendRef -> do - filePath <- lift $ toFSFilePath fPath - unless (any (`isSuffixOf` map toLower fPath) imageExtensions) $ throwChatError CEFileImageType {filePath} - fileSize <- getFileSize filePath - unless (fileSize <= maxImageSize) $ throwChatError CEFileImageSize {filePath} - -- TODO include file description for preview - processChatCommand cxt nm $ APISendMessages sendRef False Nothing False [composedMessage (Just f) (MCImage "" fixedImagePreview)] + filePath <- lift $ toFSFilePath fPath + unless (any (`isSuffixOf` map toLower fPath) imageExtensions) $ throwChatError CEFileImageType {filePath} + fileSize <- getFileSize filePath + unless (fileSize <= maxImageSize) $ throwChatError CEFileImageSize {filePath} + -- TODO include file description for preview + let cm = composedMessage (Just f) (MCImage "" fixedImagePreview) + case chatRef of + ChatRef CTFeed feedId _ -> processChatCommand cxt nm $ APISendFeedMessage feedId cm + _ -> withSendRef user chatRef $ \sendRef -> processChatCommand cxt nm $ APISendMessages sendRef False Nothing False [cm] ForwardFile chatName fileId -> forwardFile chatName fileId SendFile ForwardImage chatName fileId -> forwardFile chatName fileId SendImage SendFileDescription _chatName _f -> throwCmdError "TODO" @@ -4208,7 +4218,7 @@ processChatCommand cxt nm = \case assertUserGroupRole gInfo GRModerator let msgMemIds = itemsMsgMemIds gInfo items -- moderation deletes always sign (attributable; avoids the catch-up-moderator divergence) - signedEvents = L.nonEmpty $ map (\(msgId, memId) -> let evt = XMsgDel msgId memId (toMsgScope gInfo <$> chatScopeInfo) False in (groupMsgSigning True gInfo evt, evt)) msgMemIds + signedEvents = L.nonEmpty $ map (\(msgId, memId) -> let evt = XMsgDel msgId memId (toMsgScope gInfo <$> chatScopeInfo) False Nothing in (groupMsgSigning True gInfo evt, evt)) msgMemIds mapM_ (sendGroupSignedMessages_ gInfo ms) signedEvents delGroupChatItems user gInfo chatScopeInfo items True where @@ -5344,6 +5354,7 @@ cleanupManager = do cleanupMessages `catchAllErrors` eToView cleanupDeliveryTasks `catchAllErrors` eToView cleanupDeliveryJobs `catchAllErrors` eToView + cleanupFeedJobs `catchAllErrors` eToView -- TODO possibly, also cleanup async commands cleanupProbes `catchAllErrors` eToView liftIO $ threadDelay' $ diffToMicroseconds interval @@ -5411,6 +5422,10 @@ cleanupManager = do ts <- liftIO getCurrentTime let cutoffTs = addUTCTime (-(7 * nominalDay)) ts withStore' (`deleteDoneDeliveryJobs` cutoffTs) + cleanupFeedJobs = do + ts <- liftIO getCurrentTime + let cutoffTs = addUTCTime (-(7 * nominalDay)) ts + withStore' (`deleteDoneFeedJobs` cutoffTs) cleanupProbes = do ts <- liftIO getCurrentTime let cutoffTs = addUTCTime (-(14 * nominalDay)) ts @@ -6085,7 +6100,7 @@ chatCommandP = CTLocal -> pure $ ChatName CTLocal "" CTFeed -> pure $ ChatName CTFeed "" ct -> ChatName ct <$> displayNameP - chatNameP' = ChatName <$> (chatTypeP <|> pure CTDirect) <*> displayNameP + chatNameP' = chatNameP <|> (ChatName CTDirect <$> displayNameP) chatRefP = do chatTypeP >>= \case CTGroup -> ChatRef CTGroup <$> A.decimal <*> optional gcScopeP diff --git a/src/Simplex/Chat/Library/Internal.hs b/src/Simplex/Chat/Library/Internal.hs index e6877df732..f83e7733b9 100644 --- a/src/Simplex/Chat/Library/Internal.hs +++ b/src/Simplex/Chat/Library/Internal.hs @@ -66,6 +66,7 @@ import Simplex.Chat.Messages.CIContent.Events import Simplex.Chat.Operators import Simplex.Chat.ProfileGenerator (generateRandomProfile) import Simplex.Chat.Protocol +import Simplex.Chat.Delivery (FeedJobAction (..)) import Simplex.Chat.Store import Simplex.Chat.Store.ContactRequest import Simplex.Chat.Store.Direct @@ -1510,7 +1511,7 @@ sendHistory user gInfo@GroupInfo {membership} m@GroupMember {activeConn = Just c (Just (_, fileDescrText, fileExpires), Just msgId) -> do partSize <- asks $ xftpDescrPartSize . config let parts = splitFileDescr partSize fileDescrText - pure . L.toList $ L.map (\fd -> XMsgFileDescr msgId fd fileExpires) parts + pure . L.toList $ L.map (\fd -> XMsgFileDescr msgId fd fileExpires Nothing) parts _ -> pure [] let fileDescrVMs = map (VMUnsigned . ChatMessage senderVRange Nothing) fileDescrEvents pure $ map ((,) fwd) (contentVM : fileDescrVMs) @@ -2273,18 +2274,27 @@ createSndMessages = createSndMessages_ Nothing 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)) + mkMsg <- feedMessageMaker createdAt <- liftIO getCurrentTime - withStore' $ \db -> insertChatItemMessage_ db feedItemId (msgId' msg) createdAt - pure msg + withStore $ \db -> mkMsg db feed sharedMsgId_ feedItemId event createdAt -createFeedMessages :: Feed -> ChatItemId -> NonEmpty (ChatMsgEvent 'Json) -> CM [SndMessage] -createFeedMessages feed feedItemId events = do - (errs, msgs) <- lift $ partitionEithers . L.toList <$> createSndMessages (L.map (\evt -> (FeedId (feedId' feed), Nothing, evt)) events) - unless (null errs) $ toView $ CEvtChatErrors errs +feedMessageMaker :: CM (DB.Connection -> Feed -> Maybe SharedMsgId -> ChatItemId -> ChatMsgEvent 'Json -> UTCTime -> ExceptT StoreError IO SndMessage) +feedMessageMaker = do + g <- asks random + vr <- chatVersionRange + pure $ \db feed sharedMsgId_ feedItemId event createdAt -> do + let encodeMessage smId = encodeChatMessage maxEncodedMsgLength ChatMessage {chatVRange = vr, msgId = Just smId, chatMsgEvent = event} + msg <- createNewSndMessage db g (FeedId (feedId' feed)) sharedMsgId_ event Nothing encodeMessage + liftIO $ insertChatItemMessage_ db feedItemId (msgId' msg) createdAt + pure msg + +createFeedFileDescrJobs :: Feed -> ChatItemId -> NonEmpty (ChatMsgEvent 'Json) -> CM () +createFeedFileDescrJobs feed feedItemId events = do + mkMsg <- feedMessageMaker createdAt <- liftIO getCurrentTime - withStore' $ \db -> forM_ msgs $ \msg -> insertChatItemMessage_ db feedItemId (msgId' msg) createdAt - pure msgs + withStore $ \db -> do + msgs <- mapM (\evt -> mkMsg db feed Nothing feedItemId evt createdAt) events + liftIO $ Store.createFeedJobs db (feedId' feed) feedItemId $ FJAFileDescr (L.map msgId' msgs) feedId' :: Feed -> FeedId feedId' Feed {feedId} = feedId diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index cdadcbb0b3..dbf16bfff2 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -270,10 +270,8 @@ processAgentMsgSndFile _corrId aFileId msg = do let feedItemId = fileItemId partSize <- asks $ xftpDescrPartSize . config let parts = splitFileDescr partSize (fileDescrText rfd) - events = L.map (\fileDescr -> XMsgFileDescr {msgId = sharedMsgId, fileDescr, fileExpires}) parts - msgs <- createFeedMessages feed feedItemId events - forM_ (L.nonEmpty $ map msgId' msgs) $ \msgIds -> - withStore' $ \db -> createFeedJobs db feedId feedItemId (FJAFileDescr msgIds) + events = L.map (\fileDescr -> XMsgFileDescr {msgId = sharedMsgId, fileDescr, fileExpires, feed = Just True}) parts + createFeedFileDescrJobs feed feedItemId events startFeedWorkers feedId ci' <- withStore $ \db -> do liftIO $ updateCIFileStatus db user fileId CIFSSndComplete @@ -311,7 +309,7 @@ processAgentMsgSndFile _corrId aFileId msg = do where splitText :: (Connection, SndFileTransfer, RcvFileDescrText) -> [(Connection, (ConnOrGroupId, Maybe MsgSigning, ChatMsgEvent 'Json))] splitText (conn, _, rfdText) = - map (\fileDescr -> (conn, (connOrGroupId, Nothing, XMsgFileDescr {msgId = sharedMsgId, fileDescr, fileExpires}))) (L.toList $ splitFileDescr partSize rfdText) + map (\fileDescr -> (conn, (connOrGroupId, Nothing, XMsgFileDescr {msgId = sharedMsgId, fileDescr, fileExpires, feed = Nothing}))) (L.toList $ splitFileDescr partSize rfdText) toMsgReq :: (Connection, (ConnOrGroupId, Maybe MsgSigning, ChatMsgEvent 'Json)) -> SndMessage -> ChatMsgReq toMsgReq (conn, _) SndMessage {msgId, msgBody} = (conn, MsgFlags {notification = hasNotification XMsgFileDescr_}, (vrValue msgBody, [msgId])) @@ -566,9 +564,9 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = let ct'' = ct' {activeConn = Just conn''} :: Contact case event of XMsgNew mc -> newContentMessage ct'' mc msg msgMeta - XMsgFileDescr sharedMsgId fileDescr fileExpires -> messageFileDescription ct'' sharedMsgId fileDescr fileExpires + XMsgFileDescr sharedMsgId fileDescr fileExpires feed -> messageFileDescription ct'' sharedMsgId fileDescr fileExpires feed XMsgUpdate sharedMsgId mContent _ ttl live _msgScope _ feed -> messageUpdate ct'' sharedMsgId mContent msg msgMeta ttl live feed - XMsgDel sharedMsgId _ _ _ -> messageDelete ct'' sharedMsgId msg msgMeta + XMsgDel sharedMsgId _ _ _ feed -> messageDelete ct'' sharedMsgId msg msgMeta feed XMsgReact sharedMsgId _ _ reaction add -> directMsgReaction ct'' sharedMsgId reaction add msg msgMeta -- TODO discontinue XFile XFile fInv -> processFileInvitation' ct'' fInv msg msgMeta @@ -1056,13 +1054,13 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = where MsgContainer {scope, asGroup} = mc -- file description is always allowed, to allow sending files to support scope - XMsgFileDescr sharedMsgId fileDescr fileExpires -> groupMessageFileDescription gInfo' (Just m'') sharedMsgId fileDescr fileExpires + XMsgFileDescr sharedMsgId fileDescr fileExpires feed -> groupMessageFileDescription gInfo' (Just m'') sharedMsgId fileDescr fileExpires feed XMsgUpdate sharedMsgId mContent mentions ttl live msgScope asGroup_ feed -> checkSendAsGroup asGroup_ $ memberCanSend (Just m'') msgScope $ groupMessageUpdate gInfo' (Just m'') sharedMsgId mContent mentions msgScope msg brokerTs ttl live asGroup_ feed - XMsgDel sharedMsgId memberId_ scope_ onlyHistory -> - groupMessageDelete gInfo' (Just m'') sharedMsgId memberId_ scope_ onlyHistory msg brokerTs + XMsgDel sharedMsgId memberId_ scope_ onlyHistory feed -> + groupMessageDelete gInfo' (Just m'') sharedMsgId memberId_ scope_ onlyHistory msg brokerTs feed XMsgReact sharedMsgId memberId scope_ reaction add -> groupMsgReaction gInfo' m'' sharedMsgId memberId scope_ reaction add msg brokerTs -- TODO discontinue XFile XFile fInv -> Nothing <$ processGroupFileInvitation' gInfo' m'' fInv msg brokerTs @@ -1931,17 +1929,17 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = ChatConfig {autoAcceptFileSize = sz} <- asks config when (sz > fileSize) $ receiveFileEvt' user ft False Nothing Nothing >>= toView - messageFileDescription :: Contact -> SharedMsgId -> FileDescr -> Maybe UTCTime -> CM () - messageFileDescription Contact {contactId, chatSettings} sharedMsgId fileDescr fileExpires = - void $ skipDroppedFeedFile chatSettings $ do + messageFileDescription :: Contact -> SharedMsgId -> FileDescr -> Maybe UTCTime -> Maybe Bool -> CM () + messageFileDescription Contact {contactId, chatSettings} sharedMsgId fileDescr fileExpires feed_ = + void $ skipDroppedFeedFile chatSettings feed_ $ 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, chatSettings} m_ sharedMsgId fileDescr fileExpires = fmap join $ skipDroppedFeedFile chatSettings $ do + groupMessageFileDescription :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> FileDescr -> Maybe UTCTime -> Maybe Bool -> CM (Maybe DeliveryTaskContext) + groupMessageFileDescription g@GroupInfo {groupId, chatSettings} m_ sharedMsgId fileDescr fileExpires feed_ = fmap join $ skipDroppedFeedFile chatSettings feed_ $ do (fileId, aci) <- withStore $ \db -> do fileId <- getGroupFileIdBySharedMsgId db userId groupId sharedMsgId aci <- getChatItemByFileId db cxt user fileId @@ -2044,10 +2042,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = else toView $ CEvtChatItemNotChanged user (AChatItem SCTDirect SMDRcv (DirectChat ct) ci) _ -> messageError "x.msg.update: contact attempted invalid message update" - messageDelete :: Contact -> SharedMsgId -> RcvMessage -> MsgMeta -> CM () - messageDelete ct@Contact {contactId, chatSettings} sharedMsgId _rcvMessage msgMeta = do + messageDelete :: Contact -> SharedMsgId -> RcvMessage -> MsgMeta -> Maybe Bool -> CM () + messageDelete ct@Contact {contactId, chatSettings} sharedMsgId _rcvMessage msgMeta feed_ = do deleteRcvChatItem `catchCINotFound` \smId -> - unless (chatDropsFeed chatSettings) $ toView $ CEvtChatItemDeletedNotFound user ct smId + unless (dropsFeed chatSettings feed_) $ toView $ CEvtChatItemDeletedNotFound user ct smId where brokerTs = metaBrokerTs msgMeta deleteRcvChatItem = do @@ -2136,10 +2134,10 @@ 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 = + skipDroppedFeedFile :: ChatSettings -> Maybe Bool -> CM a -> CM (Maybe a) + skipDroppedFeedFile chatSettings feed_ f = (Just <$> f) `catchAllErrors` \case - ChatErrorStore (SEFileIdNotFoundBySharedMsgId _) | chatDropsFeed chatSettings -> pure Nothing + ChatErrorStore (SEFileIdNotFoundBySharedMsgId _) | dropsFeed chatSettings feed_ -> pure Nothing e -> throwError e validSender :: Maybe GroupMember -> CIDirection 'CTGroup 'MDRcv -> Bool @@ -2336,8 +2334,8 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = toView $ CEvtChatItemNotChanged user (AChatItem SCTGroup SMDRcv (GroupChat gInfo scopeInfo) ci) pure Nothing - groupMessageDelete :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> Maybe MemberId -> Maybe MsgScope -> Bool -> RcvMessage -> UTCTime -> CM (Maybe DeliveryTaskContext) - groupMessageDelete gInfo@GroupInfo {membership, chatSettings = groupChatSettings} m_ sharedMsgId sndMemberId_ scope_ onlyHistory rcvMsg brokerTs = + groupMessageDelete :: GroupInfo -> Maybe GroupMember -> SharedMsgId -> Maybe MemberId -> Maybe MsgScope -> Bool -> RcvMessage -> UTCTime -> Maybe Bool -> CM (Maybe DeliveryTaskContext) + groupMessageDelete gInfo@GroupInfo {membership, chatSettings = groupChatSettings} m_ sharedMsgId sndMemberId_ scope_ onlyHistory rcvMsg brokerTs feed_ = findItem >>= \case Right cci@(CChatItem _ ci@ChatItem {chatDir}) -> requireVerifiedDelete cci $ case (chatDir, m_) of (CIGroupRcv mem, Just m@GroupMember {memberId}) -> @@ -2368,7 +2366,7 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = let msgMemberId = fromMaybe memberId sndMemberId_ if | msgMemberId == memberId -> - unless (chatDropsFeed groupChatSettings) (messageError $ "x.msg.del: message not found, " <> tshow e) $> Nothing + unless (dropsFeed groupChatSettings feed_) (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 @@ -3933,10 +3931,10 @@ processAgentMessageConn cxt user@User {userId} corrId agentConnId agentMessage = where MsgContainer {scope} = mc -- file description is always allowed, to allow sending files to support scope - XMsgFileDescr sharedMsgId fileDescr fileExpires -> void $ groupMessageFileDescription gInfo author_ sharedMsgId fileDescr fileExpires + XMsgFileDescr sharedMsgId fileDescr fileExpires feed -> void $ groupMessageFileDescription gInfo author_ sharedMsgId fileDescr fileExpires feed XMsgUpdate sharedMsgId mContent mentions ttl live msgScope asGroup_ feed -> void $ memberCanSend author_ msgScope $ groupMessageUpdate gInfo author_ sharedMsgId mContent mentions msgScope rcvMsg msgTs ttl live asGroup_ feed - XMsgDel sharedMsgId memId scope_ _ -> void $ groupMessageDelete gInfo author_ sharedMsgId memId scope_ False rcvMsg msgTs + XMsgDel sharedMsgId memId scope_ _ feed -> void $ groupMessageDelete gInfo author_ sharedMsgId memId scope_ False rcvMsg msgTs feed XMsgReact sharedMsgId memId scope_ reaction add -> withAuthor XMsgReact_ $ \author -> void $ groupMsgReaction gInfo author sharedMsgId memId scope_ reaction add rcvMsg msgTs XFileCancel sharedMsgId -> void $ xFileCancelGroup gInfo author_ sharedMsgId XInfo p mKey -> withAuthor XInfo_ $ \author -> void $ xInfoMember gInfo author p mKey rcvMsg msgTs @@ -4126,7 +4124,7 @@ runDeliveryTaskWorker a deliveryKey Worker {doWork} = do forM_ body_ $ \body -> createMsgDeliveryJob db gInfo jobScope senderGMIds body forM_ acceptedTasks $ \t -> updateDeliveryTaskStatus db (deliveryTaskId t) DTSProcessed forM_ largeTasks $ \t -> setDeliveryTaskErrStatus db (deliveryTaskId t) "large" - when (isJust body_) . lift . void $ getDeliveryJobWorker True (uncurry DJKGroup deliveryKey) + when (isJust body_) . lift . void $ getDeliveryJobWorker True deliveryKey -- DJRelayRemoved is allowed when RSInactive - it forwards XGrpMemDel about relay's own deletion DJRelayRemoved | workerScope /= DWSGroup -> @@ -4138,26 +4136,38 @@ runDeliveryTaskWorker a deliveryKey Worker {doWork} = do withStore' $ \db -> do createMsgDeliveryJob db gInfo jobScope [senderGMId] body updateDeliveryTaskStatus db (deliveryTaskId task) DTSProcessed - lift . void $ getDeliveryJobWorker True (uncurry DJKGroup deliveryKey) + lift . void $ getDeliveryJobWorker True deliveryKey startDeliveryJobWorkers :: CM () startDeliveryJobWorkers = do workerScopes <- withStore' $ \db -> getPendingDeliveryJobScopes db lift $ forM_ workerScopes resumeDeliveryJobWork -resumeDeliveryJobWork :: DeliveryJobKey -> CM' () +resumeDeliveryJobWork :: DeliveryWorkerKey -> CM' () resumeDeliveryJobWork = void . getDeliveryJobWorker False -getDeliveryJobWorker :: Bool -> DeliveryJobKey -> CM' Worker +getDeliveryJobWorker :: Bool -> DeliveryWorkerKey -> CM' Worker getDeliveryJobWorker hasWork deliveryKey = do ws <- asks deliveryJobWorkers a <- asks smpAgent getAgentWorker "delivery_job" hasWork a deliveryKey ws $ runDeliveryJobWorker a deliveryKey +startFeedJobWorkers :: CM () +startFeedJobWorkers = do + feedKeys <- withStore' resumeFeedJobsOnStart + lift $ forM_ feedKeys $ void . getFeedJobWorker False + startFeedWorkers :: FeedId -> CM () startFeedWorkers feedId = - lift $ mapM_ (\scope -> void $ getDeliveryJobWorker True (DJKFeed feedId scope)) feedWorkerScopes + lift $ mapM_ (\scope -> void $ getFeedJobWorker True (feedId, scope)) feedWorkerScopes + +getFeedJobWorker :: Bool -> FeedJobKey -> CM' Worker +getFeedJobWorker hasWork feedKey = do + ws <- asks feedJobWorkers + a <- asks smpAgent + getAgentWorker "feed_job" hasWork a feedKey ws $ + runFeedJobWorker a feedKey -- TODO [relays] dissemination here is unsigned (relay-asserted profile). -- Future: members sign an XMember on channel join, relay stores it per @@ -4180,40 +4190,24 @@ encodeMemberNew vr gInfo member = case encodeChatMessage maxBatchElementSize cha chatMsgEvent = XGrpMemNew (memberInfo gInfo member) Nothing } -runDeliveryJobWorker :: AgentClient -> DeliveryJobKey -> Worker -> CM () +runDeliveryJobWorker :: AgentClient -> DeliveryWorkerKey -> Worker -> CM () runDeliveryJobWorker a deliveryKey Worker {doWork} = do delay <- asks $ deliveryWorkerDelay . config cxt <- chatStoreCxt - case deliveryKey of - DJKGroup groupId workerScope -> do - (user, gInfo) <- withStore $ \db -> do - user <- getUserByGroupId db groupId - gInfo <- getGroupInfo db cxt user groupId - pure (user, gInfo) - jobLoop delay $ - jobOperation (\db -> getNextDeliveryJob db (groupId, workerScope)) (processDeliveryJob cxt user gInfo workerScope) - DJKFeed feedId scope -> do - (user, feed) <- withStore $ \db -> do - user <- getUserByFeedId db feedId - feed <- getFeed db user feedId - pure (user, feed) - jobLoop delay $ - jobOperation (\db -> getNextFeedDeliveryJob db feedId scope) (processFeedJob cxt user feed scope) + (user, gInfo) <- withStore $ \db -> do + user <- getUserByGroupId db groupId + gInfo <- getGroupInfo db cxt user groupId + pure (user, gInfo) + jobWorkerLoop delay doWork $ + withWork_ a doWork (withStore' $ \db -> getNextDeliveryJob db deliveryKey) $ \job -> + processDeliveryJob cxt user gInfo job + `catchAllErrors` \e -> do + withStore' $ \db -> setDeliveryJobErrStatus db (deliveryJobId job) (tshow e) + eToView e where - jobLoop :: Int64 -> CM () -> CM () - jobLoop delay operation = forever $ do - unless (delay == 0) $ liftIO $ threadDelay' delay - lift $ waitForWork doWork - operation - jobOperation :: (DB.Connection -> IO (Either StoreError (Maybe (DeliveryJob c)))) -> (DeliveryJob c -> CM ()) -> CM () - jobOperation getJob processJob = - withWork_ a doWork (withStore' getJob) $ \job -> - processJob job - `catchAllErrors` \e -> do - withStore' $ \db -> setDeliveryJobErrStatus db (deliveryJobId job) (tshow e) - eToView e - processDeliveryJob :: StoreCxt -> User -> GroupInfo -> DeliveryWorkerScope -> DeliveryJob 'CTGroup -> CM () - processDeliveryJob cxt user gInfo workerScope job = + (groupId, workerScope) = deliveryKey + processDeliveryJob :: StoreCxt -> User -> GroupInfo -> MessageDeliveryJob -> CM () + processDeliveryJob cxt user gInfo job = case jobScopeImpliedSpec jobScope of DJDeliveryJob _includePending | not (relayServesGroup gInfo) -> do @@ -4231,7 +4225,7 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do deleteGroupConnections user gInfo True withStore' $ \db -> updateDeliveryJobStatus db jobId DJSComplete where - DeliveryJob {jobId, cursorId_ = startingCursor, jobWork = DJWGroup {jobScope, senderGMIds, body}} = job + MessageDeliveryJob {jobId, jobScope, senderGMIds, body, cursorGMId_ = startingCursor} = job singleSenderGMId_ = case senderGMIds of [s] -> Just s _ -> Nothing @@ -4388,51 +4382,75 @@ runDeliveryJobWorker a deliveryKey Worker {doWork} = do Nothing -> VRValue Nothing msgBody -- sending to one member, do not reference body Just 1 -> VRValue (Just 1) msgBody Just _ -> VRRef 1 - processFeedJob :: StoreCxt -> User -> Feed -> FeedWorkerScope -> DeliveryJob 'CTFeed -> CM () - processFeedJob cxt user feed scope job = do - bucketSize <- asks $ feedBucketSize . config +jobWorkerLoop :: Int64 -> TMVar () -> CM () -> CM () +jobWorkerLoop delay doWork operation = forever $ do + unless (delay == 0) $ liftIO $ threadDelay' delay + lift $ waitForWork doWork + operation + +runFeedJobWorker :: AgentClient -> FeedJobKey -> Worker -> CM () +runFeedJobWorker a feedKey@(feedId, scope) Worker {doWork} = do + delay <- asks $ deliveryWorkerDelay . config + cxt <- chatStoreCxt + (user, feed) <- withStore $ \db -> do + user <- getUserByFeedId db feedId + feed <- getFeed db user feedId + pure (user, feed) + bucketSize <- asks $ feedBucketSize . config + jobWorkerLoop delay doWork $ + withWork_ a doWork (withStore' $ \db -> getNextFeedJob db feedKey) $ \job -> + processFeedJob cxt user feed bucketSize job `catchAllErrors` jobError user feed job + where + jobError user feed FeedJob {feedJobId, feedItemId, feedAction} e = do + withStore' $ \db -> setFeedJobErrStatus db feedJobId (tshow e) + when (feedActionCreates feedAction) $ + setFeedItemStatus user feed feedItemId $ CISSndError $ SndErrOther $ tshow e + eToView $ chatErrorAgent $ CRITICAL True $ "feed job " <> show feedJobId <> " failed: " <> show e + processFeedJob :: StoreCxt -> User -> Feed -> Int -> FeedJob -> CM () + processFeedJob cxt user feed bucketSize FeedJob {feedJobId, feedItemId, feedAction, cursorId_} = do let msgIds = feedActionMsgIds feedAction msgs <- withStore' $ \db -> getFeedJobMessages db msgIds when (length msgs < length msgIds) $ logWarn "feed job: message not found" - bucketLoop bucketSize msgs startingCursor `catchAllErrors` \e -> do - when (feedActionCreates feedAction) $ - withStore' $ \db -> updateFeedChatItemStatus db user feedId feedItemId (CISSndError $ SndErrOther $ tshow e) - throwError e + bucketLoop msgs cursorId_ where - DeliveryJob {jobId, cursorId_ = startingCursor, jobWork = DJWFeed {feedItemId, feedAction}} = job - Feed {feedId} = feed - bucketLoop :: Int -> [SndMessage] -> Maybe Int64 -> CM () - bucketLoop bucketSize msgs cursor_ = + bucketLoop :: [SndMessage] -> Maybe Int64 -> CM () + bucketLoop msgs cursor_ = withStore' (\db -> runExceptT $ getFeedChatItem db user feedId feedItemId) >>= \case Right cci | not (itemChanged cci), Just item <- feedItemMsg cci -> - sendBucket bucketSize msgs item cursor_ >>= \case - Just cursor' -> bucketLoop bucketSize msgs (Just cursor') + sendBucket msgs item cursor_ >>= \case + Just cursor' -> bucketLoop msgs (Just cursor') Nothing -> completeJob _ -> completeJob itemChanged (CChatItem _ ChatItem {meta = CIMeta {itemDeleted}}) = isJust itemDeleted && not (feedActionDeletes feedAction) - sendBucket bucketSize msgs item cursor_ = do + sendBucket msgs item cursor_ = 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 -> withStore' $ \db -> do - updateFeedDeliveryJobCursor db jobId c + forM_ cursor' $ \c -> do + withStore' $ \db -> updateFeedJobCursor db feedJobId c when (feedActionCreates feedAction && isNothing cursor_) $ - updateFeedChatItemStatus db user feedId feedItemId (CISSndSent SSPPartial) + setFeedItemStatus user feed feedItemId $ CISSndSent SSPPartial pure cursor' - completeJob = do - lastOfEvent <- - withFeedLock "completeFeedJob" feedId $ - withStore' $ \db -> completeFeedJob db jobId feedItemId (feedActionTag feedAction) - when lastOfEvent $ finishFeedEvent user feed feedItemId feedAction + completeJob = + withFeedLock "completeFeedJob" feedId $ do + lastOfEvent <- withStore' $ \db -> completeFeedJob db feedJobId feedItemId (feedActionTag feedAction) + when lastOfEvent $ finishFeedEvent user feed feedItemId feedAction -chatDropsFeed :: ChatSettings -> Bool -chatDropsFeed ChatSettings {dropFeed} = isTrue dropFeed +setFeedItemStatus :: User -> Feed -> ChatItemId -> CIStatus 'MDSnd -> CM () +setFeedItemStatus user feed@Feed {feedId} feedItemId status = do + withStore' $ \db -> updateFeedChatItemStatus db user feedId feedItemId status + withFeedItem user feed feedItemId $ \ci -> toView $ CEvtChatItemsStatusesUpdated user [aFeedItem feed ci] + +withFeedItem :: User -> Feed -> ChatItemId -> (CChatItem 'CTFeed -> CM ()) -> CM () +withFeedItem user Feed {feedId} feedItemId action = + withStore' (\db -> runExceptT $ getFeedChatItem db user feedId feedItemId) + >>= either (const $ pure ()) action dropsFeed :: ChatSettings -> Maybe Bool -> Bool -dropsFeed chatSettings feed_ = feed_ == Just True && chatDropsFeed chatSettings +dropsFeed ChatSettings {dropFeed} feed_ = feed_ == Just True && isTrue dropFeed dropsFeedMsg :: ChatSettings -> MsgContainer -> Bool dropsFeedMsg chatSettings MsgContainer {feed} = dropsFeed chatSettings feed @@ -4449,14 +4467,8 @@ feedContactsBucket cxt user feedItemId action item bucketSize msgs cursor_ = cas forM_ (lastId ctIds) $ \lastCtId -> do existing <- withStore' $ \db -> getFeedInstanceContactIdsByRange db user feedItemId fromId lastCtId createdAt <- liftIO getCurrentTime - recipients <- withStore' $ \db -> - forM (mapMaybe feedRecipient cts) $ \(ct, conn) -> do - let timed_ = sndFeedTimed (contactTimedTTL ct) createdAt - case M.lookup (contactId' ct) existing of - Just itemId -> pure (contactId' ct, itemId, conn, Nothing) - Nothing -> do - itemId <- createFeedInstanceItem db user (CDDirectSnd ct) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt - pure (contactId' ct, itemId, conn, timed_) + (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db createdAt existing) (mapMaybe feedRecipient cts)) + unless (null errs) $ toView $ CEvtChatErrors errs forM_ recipients $ \(ctId, itemId, _, newTimed_) -> forM_ (newTimed_ >>= timedDeleteAt') $ startProximateTimedItemThread user (ChatRef CTDirect ctId Nothing, itemId) @@ -4477,6 +4489,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 + instanceRecipient db createdAt existing (ct, conn) = + let timed_ = sndFeedTimed (contactTimedTTL ct) createdAt + in case M.lookup (contactId' ct) existing of + Just itemId -> pure $ Right (contactId' ct, itemId, conn, Nothing) + Nothing -> do + itemId <- createFeedInstanceItem db user (CDDirectSnd ct) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt + pure $ Right (contactId' ct, itemId, conn, timed_) deliverBucket lastCtId recipients = do 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] @@ -4510,21 +4529,14 @@ feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case members <- withStore' $ \db -> getCustomerGroupsMembersByRange db cxt user fromId lastGId existing <- withStore' $ \db -> getFeedInstanceGroupIdsByRange db user feedItemId fromId lastGId createdAt <- liftIO getCurrentTime - recipients <- withStore' $ \db -> - forM (filter feedGroup gs) $ \g -> do - let ms = groupMembers members g - timed_ = sndFeedTimed (groupTimedTTL g) createdAt - case M.lookup (groupId' g) existing of - Just itemId -> pure (groupId' g, itemId, ms, Nothing) - Nothing -> do - itemId <- createFeedInstanceItem db user (CDGroupSnd g Nothing) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt - forM_ ms $ \(m, _) -> createGroupSndStatus db itemId (groupMemberId' m) GSSNew - pure (groupId' g, itemId, ms, timed_) + (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db members createdAt existing) (filter feedGroup gs)) + unless (null errs) $ toView $ CEvtChatErrors errs forM_ recipients $ \(gId, itemId, _, newTimed_) -> forM_ (newTimed_ >>= timedDeleteAt') $ startProximateTimedItemThread user (ChatRef CTGroup gId Nothing, itemId) results <- deliverBucket lastGId [(itemId, ms) | (_, itemId, ms, _) <- recipients] - withStore' $ \db -> updateFeedInstanceStatuses db (sendErrorStatuses results) + withStore' $ \db -> forM_ results $ \((itemId, mId), r) -> + forM_ (resultSndError r) $ \e -> updateGroupSndStatus db itemId mId (GSSError e) pure $ bucketCursor bucketSize gIds _ -> do instances <- withStore' $ \db -> getFeedGroupInstancesByCursor db cxt user feedItemId (feedActionInstances action) cursor_ bucketSize @@ -4542,9 +4554,19 @@ 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]] + instanceRecipient db members createdAt existing g = + let ms = groupMembers members g + timed_ = sndFeedTimed (groupTimedTTL g) createdAt + in case M.lookup (groupId' g) existing of + Just itemId -> pure $ Right (groupId' g, itemId, ms, Nothing) + Nothing -> do + itemId <- createFeedInstanceItem db user (CDGroupSnd g Nothing) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt + forM_ ms $ \(m, _) -> createGroupSndStatus db itemId (groupMemberId' m) GSSNew + pure $ Right (groupId' g, itemId, ms, timed_) deliverBucket lastGId recipients = do 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] + let toSend = [((itemId, groupMemberId' m), conn) | (itemId, ms) <- recipients, (m, conn) <- ms, not $ groupMemberId' m `S.member` delivered] + deliverFeedBucket action msgs toSend applyGroupAction :: User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> [(GroupInfo, ChatItemId)] -> GroupId -> GroupId -> CM () applyGroupAction user feedItemId action item instances fromId toId = case action of @@ -4566,26 +4588,29 @@ 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) -deliverFeedBucket :: FeedJobAction -> [SndMessage] -> [(ChatItemId, Connection)] -> CM [(ChatItemId, Either ChatError ([Int64], PQEncryption))] +deliverFeedBucket :: FeedJobAction -> [SndMessage] -> [(r, Connection)] -> CM [(r, 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 $ feedActionEventTag action} - (itemIds, msgReqs) = sharedBodyReqs msgFlags (length batched' + length msgs') msgBatchMBR batched' recipients + let msgFlags = MsgFlags {notification = feedActionCreates action} + (rs, msgReqs) = sharedBodyReqs msgFlags (length batched' + length msgs') msgBatchMBR batched' recipients case L.nonEmpty msgReqs of Nothing -> pure [] - Just msgReqs' -> zip itemIds . L.toList <$> deliverMessagesB msgReqs' + Just msgReqs' -> zip rs . L.toList <$> deliverMessagesB msgReqs' _ -> pure [] +resultSndError :: Either ChatError a -> Maybe SndError +resultSndError = \case + Left e -> Just $ SndErrOther $ tshow e + Right _ -> Nothing + sendErrorStatuses :: [(ChatItemId, Either ChatError a)] -> [(ChatItemId, CIStatus 'MDSnd)] sendErrorStatuses = mapMaybe itemError where - itemError (itemId, r) = case r of - Left e -> Just (itemId, CISSndError $ SndErrOther $ tshow e) - Right _ -> Nothing + itemError (itemId, r) = (itemId,) . CISSndError <$> resultSndError r firstMsgId :: [SndMessage] -> MessageId firstMsgId = \case @@ -4604,18 +4629,12 @@ bucketCursor bucketSize ids finishFeedEvent :: User -> Feed -> ChatItemId -> FeedJobAction -> CM () finishFeedEvent user feed@Feed {feedId} feedItemId action - | feedActionCreates action = do - withStore' $ \db -> updateFeedChatItemStatus db user feedId feedItemId (CISSndSent SSPComplete) - withFeedItem $ \ci -> toView $ CEvtChatItemsStatusesUpdated user [aFeedItem feed ci] - | feedActionRemovesItem action = withFeedItem $ \cci -> do + | feedActionCreates action = setFeedItemStatus user feed feedItemId $ CISSndSent SSPComplete + | feedActionRemovesItem action = withFeedItem user feed feedItemId $ \cci -> do deleteCIFiles user $ itemsFilesInfo [cci] withStore' $ \db -> deleteFeedChatItem db user feedId feedItemId toView $ CEvtChatItemsDeleted user [ChatItemDeletion (aFeedItem feed cci) Nothing] True False | otherwise = pure () - where - withFeedItem 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. diff --git a/src/Simplex/Chat/Protocol.hs b/src/Simplex/Chat/Protocol.hs index 7882e0a030..f597418805 100644 --- a/src/Simplex/Chat/Protocol.hs +++ b/src/Simplex/Chat/Protocol.hs @@ -451,9 +451,9 @@ signChatMsgBody MsgSigning {bindingTag, bindingData, keyRef, privKey} msgBody = data ChatMsgEvent (e :: MsgEncoding) where XMsgNew :: MsgContainer -> ChatMsgEvent 'Json - XMsgFileDescr :: {msgId :: SharedMsgId, fileDescr :: FileDescr, fileExpires :: Maybe UTCTime} -> ChatMsgEvent 'Json + XMsgFileDescr :: {msgId :: SharedMsgId, fileDescr :: FileDescr, fileExpires :: Maybe UTCTime, feed :: Maybe Bool} -> ChatMsgEvent 'Json XMsgUpdate :: {msgId :: SharedMsgId, content :: MsgContent, mentions :: Map MemberName MsgMention, ttl :: Maybe Int, live :: Maybe Bool, scope :: Maybe MsgScope, asGroup :: Maybe Bool, feed :: Maybe Bool} -> ChatMsgEvent 'Json - XMsgDel :: {msgId :: SharedMsgId, memberId :: Maybe MemberId, scope :: Maybe MsgScope, onlyHistory :: Bool} -> ChatMsgEvent 'Json + XMsgDel :: {msgId :: SharedMsgId, memberId :: Maybe MemberId, scope :: Maybe MsgScope, onlyHistory :: Bool, feed :: Maybe Bool} -> ChatMsgEvent 'Json XMsgDeleted :: ChatMsgEvent 'Json XMsgReact :: {msgId :: SharedMsgId, memberId :: Maybe MemberId, scope :: Maybe MsgScope, reaction :: MsgReaction, add :: Bool} -> ChatMsgEvent 'Json XFile :: FileInvitation -> ChatMsgEvent 'Json -- TODO discontinue @@ -1404,7 +1404,7 @@ appJsonToCM AppMessageJson {v, msgId, event, params} = do msg :: CMEventTag 'Json -> Either String (ChatMsgEvent 'Json) msg = \case XMsgNew_ -> XMsgNew <$> JT.parseEither parseJSON (J.Object params) - XMsgFileDescr_ -> XMsgFileDescr <$> p "msgId" <*> p "fileDescr" <*> opt "fileExpires" + XMsgFileDescr_ -> XMsgFileDescr <$> p "msgId" <*> p "fileDescr" <*> opt "fileExpires" <*> opt "feed" XMsgUpdate_ -> do msgId' <- p "msgId" content <- p "content" @@ -1415,7 +1415,7 @@ appJsonToCM AppMessageJson {v, msgId, event, params} = do asGroup <- opt "asGroup" feed <- opt "feed" pure XMsgUpdate {msgId = msgId', content, mentions, ttl, live, scope, asGroup, feed} - XMsgDel_ -> XMsgDel <$> p "msgId" <*> opt "memberId" <*> opt "scope" <*> (fromMaybe False <$> opt "onlyHistory") + XMsgDel_ -> XMsgDel <$> p "msgId" <*> opt "memberId" <*> opt "scope" <*> (fromMaybe False <$> opt "onlyHistory") <*> opt "feed" XMsgDeleted_ -> pure XMsgDeleted XMsgReact_ -> XMsgReact <$> p "msgId" <*> opt "memberId" <*> opt "scope" <*> p "reaction" <*> p "add" XFile_ -> XFile <$> p "file" @@ -1497,9 +1497,9 @@ chatToAppMessage chatMsg@ChatMessage {chatVRange, msgId, chatMsgEvent} = case en XMsgNew mc -> case toJSON mc of J.Object obj -> obj _ -> JM.empty - XMsgFileDescr msgId' fileDescr fileExpires -> o $ ("fileExpires" .=? fileExpires) ["msgId" .= msgId', "fileDescr" .= fileDescr] + XMsgFileDescr msgId' fileDescr fileExpires feed -> o $ ("feed" .=? feed) $ ("fileExpires" .=? fileExpires) ["msgId" .= msgId', "fileDescr" .= fileDescr] XMsgUpdate {msgId = msgId', content, mentions, ttl, live, scope, asGroup, feed} -> o $ ("feed" .=? feed) $ ("asGroup" .=? asGroup) $ ("ttl" .=? ttl) $ ("live" .=? live) $ ("scope" .=? scope) $ ("mentions" .=? nonEmptyMap mentions) ["msgId" .= msgId', "content" .= content] - XMsgDel msgId' memberId scope onlyHistory -> o $ ("memberId" .=? memberId) $ ("scope" .=? scope) $ ("onlyHistory" .=? justTrue onlyHistory) ["msgId" .= msgId'] + XMsgDel msgId' memberId scope onlyHistory feed -> o $ ("feed" .=? feed) $ ("memberId" .=? memberId) $ ("scope" .=? scope) $ ("onlyHistory" .=? justTrue onlyHistory) ["msgId" .= msgId'] XMsgDeleted -> JM.empty XMsgReact msgId' memberId scope reaction add -> o $ ("memberId" .=? memberId) $ ("scope" .=? scope) ["msgId" .= msgId', "reaction" .= reaction, "add" .= add] XFile fileInv -> o ["file" .= fileInv] diff --git a/src/Simplex/Chat/Store/Delivery.hs b/src/Simplex/Chat/Store/Delivery.hs index a4eaacfb57..f70cae4ead 100644 --- a/src/Simplex/Chat/Store/Delivery.hs +++ b/src/Simplex/Chat/Store/Delivery.hs @@ -26,11 +26,6 @@ module Simplex.Chat.Store.Delivery getGroupMembersByCursor, updateDeliveryJobCursor, deleteDoneDeliveryJobs, - createFeedJobs, - getNextFeedDeliveryJob, - updateFeedDeliveryJobCursor, - completeFeedJob, - getFeedJobMessages, ) where @@ -38,13 +33,11 @@ import qualified Data.Aeson as J 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) +import Data.Maybe (isNothing) import Data.Text (Text) import qualified Data.Text as T import Data.Time.Clock (UTCTime, getCurrentTime) import Simplex.Chat.Delivery -import Simplex.Chat.Messages (ChatItemId, ChatType (..), MessageId, SndMessage (..)) import Simplex.Chat.Protocol hiding (Binary) import Simplex.Chat.Store.Shared import Simplex.Chat.Types @@ -276,33 +269,22 @@ createMsgDeliveryJob db gInfo jobScope senderGMIds body = do | null senderGMIds = Nothing | otherwise = Just $ T.intercalate "," $ map (T.pack . show) senderGMIds -getPendingDeliveryJobScopes :: DB.Connection -> IO [DeliveryJobKey] -getPendingDeliveryJobScopes db = do - groupScopes <- - DB.query - db - [sql| - SELECT DISTINCT group_id, worker_scope - FROM delivery_jobs - WHERE failed = 0 AND job_status = ? AND group_id IS NOT NULL - |] - (Only DJSPending) - 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 +getPendingDeliveryJobScopes :: DB.Connection -> IO [DeliveryWorkerKey] +getPendingDeliveryJobScopes db = + DB.query + db + [sql| + SELECT DISTINCT group_id, worker_scope + FROM delivery_jobs + WHERE failed = 0 AND job_status = ? + |] + (Only DJSPending) type MessageDeliveryJobRow = (Only Int64) :. DeliveryJobScopeRow :. (Maybe Text, Binary ByteString, Maybe GroupMemberId) -getNextDeliveryJob :: DB.Connection -> DeliveryWorkerKey -> IO (Either StoreError (Maybe (DeliveryJob 'CTGroup))) +getNextDeliveryJob :: DB.Connection -> DeliveryWorkerKey -> IO (Either StoreError (Maybe MessageDeliveryJob)) getNextDeliveryJob db deliveryKey = do - getWorkItem "delivery job" getJobId getJob (markJobFailed db) + getWorkItem "delivery job" getJobId getJob markJobFailed where (groupId, workerScope) = deliveryKey getJobId :: IO (Maybe Int64) @@ -319,7 +301,7 @@ getNextDeliveryJob db deliveryKey = do LIMIT 1 |] (groupId, workerScope, DJSPending) - getJob :: Int64 -> IO (Either StoreError (DeliveryJob 'CTGroup)) + getJob :: Int64 -> IO (Either StoreError MessageDeliveryJob) getJob jobId = firstRow' toDeliveryJob (SEDeliveryJobNotFound jobId) $ DB.query @@ -334,123 +316,14 @@ getNextDeliveryJob db deliveryKey = do |] (Only jobId) where - toDeliveryJob :: MessageDeliveryJobRow -> Either StoreError (DeliveryJob 'CTGroup) - toDeliveryJob ((Only jobId') :. jobScopeRow :. (senderGMIdsText_, Binary body, cursorId_)) = do + toDeliveryJob :: MessageDeliveryJobRow -> Either StoreError MessageDeliveryJob + toDeliveryJob ((Only jobId') :. jobScopeRow :. (senderGMIdsText_, Binary body, cursorGMId_)) = do jobScope <- maybe (Left $ SEInvalidDeliveryJob jobId') Right $ toJobScope_ jobScopeRow - senderGMIds <- parseIds jobId' senderGMIdsText_ - Right DeliveryJob {jobId = jobId', cursorId_, jobWork = DJWGroup {jobScope, senderGMIds, body}} - -parseIds :: Int64 -> Maybe Text -> Either StoreError [Int64] -parseIds jobId = \case - Nothing -> Right [] - Just t - | T.null t -> Right [] - | otherwise -> maybe (Left $ SEInvalidDeliveryJob jobId) Right $ traverse (readMaybe . T.unpack) (T.splitOn "," t) - -idsColumn :: [Int64] -> Maybe Text -idsColumn ids - | null ids = Nothing - | otherwise = Just $ T.intercalate "," $ map (T.pack . show) ids - -markJobFailed :: DB.Connection -> Int64 -> IO () -markJobFailed db jobId = - DB.execute db "UPDATE delivery_jobs SET failed = 1 where delivery_job_id = ?" (Only jobId) - -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 ( - feed_id, chat_item_id, worker_scope, job_scope_spec_tag, message_ids, - job_status, created_at, updated_at - ) VALUES (?,?,?,?,?,?,?,?) - |] - [(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) - -getNextFeedDeliveryJob :: DB.Connection -> FeedId -> FeedWorkerScope -> IO (Either StoreError (Maybe (DeliveryJob 'CTFeed))) -getNextFeedDeliveryJob db feedId scope = do - getWorkItem "feed delivery job" getJobId getJob (markJobFailed db) - where - getJobId :: IO (Maybe Int64) - getJobId = - maybeFirstRow fromOnly $ - DB.query - db - [sql| - SELECT delivery_job_id - FROM delivery_jobs - WHERE feed_id = ? AND worker_scope = ? - AND failed = 0 AND job_status = ? - ORDER BY delivery_job_id ASC - LIMIT 1 - |] - (feedId, scope, DJSPending) - getJob :: Int64 -> IO (Either StoreError (DeliveryJob 'CTFeed)) - getJob jobId = - firstRow' toFeedDeliveryJob (SEDeliveryJobNotFound jobId) $ - DB.query - db - [sql| - SELECT delivery_job_id, chat_item_id, job_scope_spec_tag, message_ids, feed_cursor_id - FROM delivery_jobs - WHERE delivery_job_id = ? - |] - (Only jobId) - toFeedDeliveryJob :: FeedDeliveryJobRow -> Either StoreError (DeliveryJob 'CTFeed) - toFeedDeliveryJob (jobId, feedItemId, actionTag, msgIdsText_, cursorId_) = do - msgIds <- parseIds jobId msgIdsText_ - feedAction <- case (actionTag, msgIds) of - (FJATNew, [msgId]) -> Right $ FJANew msgId - (FJATFileDescr, msgId : msgIds') -> Right $ FJAFileDescr (msgId :| msgIds') - (FJATUpdate, [msgId]) -> Right $ FJAUpdate msgId - (FJATDeleteBroadcast, [msgId]) -> Right $ FJADeleteBroadcast msgId - (FJATDeleteInternal, []) -> Right FJADeleteInternal - (FJATDeleteMark, []) -> Right FJADeleteMark - _ -> Left $ SEInvalidDeliveryJob jobId - Right DeliveryJob {jobId, cursorId_, jobWork = DJWFeed {feedItemId, feedAction}} - -updateFeedDeliveryJobCursor :: DB.Connection -> Int64 -> Int64 -> IO () -updateFeedDeliveryJobCursor db jobId cursorId = do - currentTs <- getCurrentTime - DB.execute - db - "UPDATE delivery_jobs SET feed_cursor_id = ?, updated_at = ? WHERE delivery_job_id = ?" - (cursorId, currentTs, jobId) - -completeFeedJob :: DB.Connection -> Int64 -> ChatItemId -> FeedJobActionTag -> IO Bool -completeFeedJob db jobId feedItemId actionTag = do - updateDeliveryJobStatus db jobId DJSComplete - 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 != ? - |] - (feedItemId, actionTag, DJSComplete) - pure $ unfinished == Just (0 :: Int) - -getFeedJobMessages :: DB.Connection -> [MessageId] -> IO [SndMessage] -getFeedJobMessages db = fmap catMaybes . mapM getMsg - where - getMsg msgId = - maybeFirstRow toSndMessage $ - DB.query db "SELECT shared_msg_id, msg_body FROM messages WHERE message_id = ? AND shared_msg_id IS NOT NULL" (Only msgId) - where - toSndMessage (sharedMsgId, Binary msgBody) = SndMessage {msgId, sharedMsgId, msgBody, signedMsg_ = Nothing} + senderGMIds <- maybe (Left $ SEInvalidDeliveryJob jobId') Right $ parseIds senderGMIdsText_ + Right $ MessageDeliveryJob {jobId = jobId', jobScope, senderGMIds, body, cursorGMId_} + markJobFailed :: Int64 -> IO () + markJobFailed jobId = + DB.execute db "UPDATE delivery_jobs SET failed = 1 where delivery_job_id = ?" (Only jobId) updateDeliveryJobStatus :: DB.Connection -> Int64 -> DeliveryJobStatus -> IO () updateDeliveryJobStatus db jobId status = updateDeliveryJobStatus_ db jobId status Nothing diff --git a/src/Simplex/Chat/Store/Direct.hs b/src/Simplex/Chat/Store/Direct.hs index 9f49723ada..736074d4af 100644 --- a/src/Simplex/Chat/Store/Direct.hs +++ b/src/Simplex/Chat/Store/Direct.hs @@ -666,6 +666,7 @@ setUserChatsRead db User {userId} = do DB.execute db "UPDATE contacts SET unread_chat = ?, updated_at = ? WHERE user_id = ? AND unread_chat = ?" (BI False, updatedAt, userId, BI True) DB.execute db "UPDATE groups SET unread_chat = ?, updated_at = ? WHERE user_id = ? AND unread_chat = ?" (BI False, updatedAt, userId, BI True) DB.execute db "UPDATE note_folders SET unread_chat = ?, updated_at = ? WHERE user_id = ? AND unread_chat = ?" (BI False, updatedAt, userId, BI True) + DB.execute db "UPDATE feeds SET unread_chat = ?, updated_at = ? WHERE user_id = ? AND unread_chat = ?" (BI False, updatedAt, userId, BI True) DB.execute db "UPDATE chat_items SET item_status = ?, item_viewed = 1, updated_at = ? WHERE user_id = ? AND item_status = ?" (CISRcvRead, updatedAt, userId, CISRcvNew) updateContactStatus :: DB.Connection -> User -> Contact -> ContactStatus -> IO Contact diff --git a/src/Simplex/Chat/Store/Feeds.hs b/src/Simplex/Chat/Store/Feeds.hs index d728b56282..4b42e38aab 100644 --- a/src/Simplex/Chat/Store/Feeds.hs +++ b/src/Simplex/Chat/Store/Feeds.hs @@ -33,6 +33,14 @@ module Simplex.Chat.Store.Feeds detachFeedInstances, deleteFeedContactReactions, deleteFeedGroupReactions, + createFeedJobs, + resumeFeedJobsOnStart, + getNextFeedJob, + updateFeedJobCursor, + setFeedJobErrStatus, + completeFeedJob, + getFeedJobMessages, + deleteDoneFeedJobs, feedItemMsg, FeedItemMsg (..), ) @@ -42,19 +50,21 @@ import Control.Monad (forM_, unless) import Control.Monad.Except (ExceptT (..), throwError) import Control.Monad.IO.Class (liftIO) import Data.Int (Int64) +import Data.List.NonEmpty (NonEmpty (..)) import Data.Map.Strict (Map) import qualified Data.Map.Strict as M -import Data.Maybe (fromMaybe) +import Data.Maybe (catMaybes, fromMaybe) import Data.Text (Text) import Data.Time (UTCTime, getCurrentTime) -import Simplex.Chat.Delivery (FeedInstanceSpec (..)) +import Simplex.Chat.Delivery import Simplex.Chat.Messages import Simplex.Chat.Messages.CIContent import Simplex.Chat.Store.Shared import Simplex.Chat.Types -import Simplex.Messaging.Agent.Store.AgentStore (firstRow) -import Simplex.Messaging.Agent.Store.DB (BoolInt (..)) +import Simplex.Messaging.Agent.Store.AgentStore (firstRow, getWorkItem, maybeFirstRow) +import Simplex.Messaging.Agent.Store.DB (Binary (..), BoolInt (..)) import qualified Simplex.Messaging.Agent.Store.DB as DB +import Simplex.Messaging.Util (firstRow') #if defined(dbPostgres) import Database.PostgreSQL.Simple (In (..), Only (..), Query, ToRow, (:.) (..)) import Database.PostgreSQL.Simple.SqlQQ (sql) @@ -262,7 +272,6 @@ updateFeedInstances_ db User {userId} feedItemId chatIdColumn fromId toId FeedIt deleteFeedInstances :: DB.Connection -> [ChatItemId] -> IO () deleteFeedInstances db itemIds = do - 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 @@ -302,3 +311,138 @@ itemIdsStmt db stmt params itemIds = unless (null itemIds) execIds cursorId :: Maybe Int64 -> Int64 cursorId = fromMaybe 0 + +createFeedJobs :: DB.Connection -> FeedId -> ChatItemId -> FeedJobAction -> IO () +createFeedJobs db feedId feedItemId action = do + currentTs <- getCurrentTime + DB.execute + db + "DELETE FROM feed_jobs WHERE chat_item_id = ? AND action_tag = ? AND (job_status = ? OR failed = 1)" + (feedItemId, tag, DJSError) + DB.executeMany + db + [sql| + INSERT INTO feed_jobs ( + feed_id, chat_item_id, worker_scope, action_tag, message_ids, + job_status, created_at, updated_at + ) VALUES (?,?,?,?,?,?,?,?) + |] + [(feedId, feedItemId, scope, tag, msgIds, DJSPending, currentTs, currentTs) | scope <- feedWorkerScopes] + where + tag = feedActionTag action + msgIds = idsColumn $ feedActionMsgIds action + +resumeFeedJobsOnStart :: DB.Connection -> IO [FeedJobKey] +resumeFeedJobsOnStart db = do + currentTs <- getCurrentTime + DB.execute + db + "UPDATE feed_jobs SET job_status = ?, updated_at = ? WHERE failed = 0 AND job_status = ?" + (DJSPending, currentTs, DJSError) + DB.query + db + [sql| + SELECT DISTINCT feed_id, worker_scope + FROM feed_jobs + WHERE failed = 0 AND job_status = ? + |] + (Only DJSPending) + +type FeedJobRow = (Int64, ChatItemId, FeedJobActionTag, Maybe Text, Maybe Int64) + +getNextFeedJob :: DB.Connection -> FeedJobKey -> IO (Either StoreError (Maybe FeedJob)) +getNextFeedJob db (feedId, scope) = + getWorkItem "feed job" getJobId getJob markJobFailed + where + getJobId :: IO (Maybe Int64) + getJobId = + maybeFirstRow fromOnly $ + DB.query + db + [sql| + SELECT feed_job_id + FROM feed_jobs + WHERE feed_id = ? AND worker_scope = ? + AND failed = 0 AND job_status = ? + ORDER BY feed_job_id ASC + LIMIT 1 + |] + (feedId, scope, DJSPending) + getJob :: Int64 -> IO (Either StoreError FeedJob) + getJob jobId = + firstRow' toFeedJob (SEFeedJobNotFound jobId) $ + DB.query + db + [sql| + SELECT feed_job_id, chat_item_id, action_tag, message_ids, cursor_id + FROM feed_jobs + WHERE feed_job_id = ? + |] + (Only jobId) + where + toFeedJob :: FeedJobRow -> Either StoreError FeedJob + toFeedJob (feedJobId, feedItemId, actionTag, msgIdsText_, cursorId_) = do + msgIds <- maybe (Left $ SEInvalidFeedJob feedJobId) Right $ parseIds msgIdsText_ + feedAction <- case (actionTag, msgIds) of + (FJATNew, [msgId]) -> Right $ FJANew msgId + (FJATFileDescr, msgId : msgIds') -> Right $ FJAFileDescr (msgId :| msgIds') + (FJATUpdate, [msgId]) -> Right $ FJAUpdate msgId + (FJATDeleteBroadcast, [msgId]) -> Right $ FJADeleteBroadcast msgId + (FJATDeleteInternal, []) -> Right FJADeleteInternal + (FJATDeleteMark, []) -> Right FJADeleteMark + _ -> Left $ SEInvalidFeedJob feedJobId + Right FeedJob {feedJobId, feedItemId, feedAction, cursorId_} + markJobFailed :: Int64 -> IO () + markJobFailed jobId = + DB.execute db "UPDATE feed_jobs SET failed = 1 WHERE feed_job_id = ?" (Only jobId) + +updateFeedJobCursor :: DB.Connection -> Int64 -> Int64 -> IO () +updateFeedJobCursor db jobId cursorId_ = do + currentTs <- getCurrentTime + DB.execute + db + "UPDATE feed_jobs SET cursor_id = ?, updated_at = ? WHERE feed_job_id = ?" + (cursorId_, currentTs, jobId) + +setFeedJobErrStatus :: DB.Connection -> Int64 -> Text -> IO () +setFeedJobErrStatus db jobId errReason = do + currentTs <- getCurrentTime + DB.execute + db + "UPDATE feed_jobs SET job_status = ?, job_err_reason = ?, updated_at = ? WHERE feed_job_id = ?" + (DJSError, errReason, currentTs, jobId) + +completeFeedJob :: DB.Connection -> Int64 -> ChatItemId -> FeedJobActionTag -> IO Bool +completeFeedJob db jobId feedItemId actionTag = do + currentTs <- getCurrentTime + DB.execute + db + "UPDATE feed_jobs SET job_status = ?, updated_at = ? WHERE feed_job_id = ?" + (DJSComplete, currentTs, jobId) + unfinished <- + maybeFirstRow fromOnly $ + DB.query + db + [sql| + SELECT COUNT(1) + FROM feed_jobs + WHERE chat_item_id = ? AND action_tag = ? AND job_status != ? + |] + (feedItemId, actionTag, DJSComplete) + pure $ unfinished == Just (0 :: Int) + +getFeedJobMessages :: DB.Connection -> [MessageId] -> IO [SndMessage] +getFeedJobMessages db = fmap catMaybes . mapM getMsg + where + getMsg msgId = + maybeFirstRow toSndMessage $ + DB.query db "SELECT shared_msg_id, msg_body FROM messages WHERE message_id = ? AND shared_msg_id IS NOT NULL" (Only msgId) + where + toSndMessage (sharedMsgId, Binary msgBody) = SndMessage {msgId, sharedMsgId, msgBody, signedMsg_ = Nothing} + +deleteDoneFeedJobs :: DB.Connection -> UTCTime -> IO () +deleteDoneFeedJobs db createdAtCutoff = + DB.execute + db + "DELETE FROM feed_jobs WHERE created_at <= ? AND (job_status = ? OR failed = 1)" + (createdAtCutoff, DJSComplete) diff --git a/src/Simplex/Chat/Store/Postgres/Migrations/M20260905_feeds.hs b/src/Simplex/Chat/Store/Postgres/Migrations/M20260905_feeds.hs index 2c262ef7ef..57630ae30e 100644 --- a/src/Simplex/Chat/Store/Postgres/Migrations/M20260905_feeds.hs +++ b/src/Simplex/Chat/Store/Postgres/Migrations/M20260905_feeds.hs @@ -41,29 +41,31 @@ ALTER TABLE groups ADD COLUMN drop_feed SMALLINT NOT NULL DEFAULT 0; CREATE INDEX idx_contacts_user_id ON contacts(user_id, contact_id); CREATE INDEX idx_groups_user_id_business_chat ON groups(user_id, business_chat, group_id); -ALTER TABLE delivery_jobs ALTER COLUMN group_id DROP NOT NULL; - -ALTER TABLE delivery_jobs ADD COLUMN feed_id BIGINT REFERENCES feeds ON DELETE CASCADE; -ALTER TABLE delivery_jobs ADD COLUMN chat_item_id BIGINT REFERENCES chat_items ON DELETE CASCADE; -ALTER TABLE delivery_jobs ADD COLUMN message_ids TEXT; -ALTER TABLE delivery_jobs ADD COLUMN feed_cursor_id BIGINT; -CREATE INDEX idx_delivery_jobs_feed_next ON delivery_jobs(feed_id, worker_scope, failed, job_status); -CREATE INDEX idx_delivery_jobs_chat_item_id ON delivery_jobs(chat_item_id); +CREATE TABLE feed_jobs ( + feed_job_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, + feed_id BIGINT NOT NULL REFERENCES feeds ON DELETE CASCADE, + chat_item_id BIGINT NOT NULL REFERENCES chat_items ON DELETE CASCADE, + worker_scope TEXT NOT NULL, + action_tag TEXT NOT NULL, + message_ids TEXT, + cursor_id BIGINT, + job_status TEXT NOT NULL, + job_err_reason TEXT, + failed SMALLINT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT (now()), + updated_at TIMESTAMPTZ NOT NULL DEFAULT (now()) +); +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); |] down_m20260905_feeds :: Text down_m20260905_feeds = T.pack [r| -DROP INDEX idx_delivery_jobs_chat_item_id; -DROP INDEX idx_delivery_jobs_feed_next; -DELETE FROM delivery_jobs WHERE group_id IS NULL; -ALTER TABLE delivery_jobs DROP COLUMN feed_cursor_id; -ALTER TABLE delivery_jobs DROP COLUMN message_ids; -ALTER TABLE delivery_jobs DROP COLUMN chat_item_id; -ALTER TABLE delivery_jobs DROP COLUMN feed_id; - -ALTER TABLE delivery_jobs ALTER COLUMN group_id SET NOT NULL; +DROP INDEX idx_feed_jobs_chat_item_id; +DROP INDEX idx_feed_jobs_next; +DROP TABLE feed_jobs; DROP INDEX idx_groups_user_id_business_chat; DROP INDEX idx_contacts_user_id; diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/M20260905_feeds.hs b/src/Simplex/Chat/Store/SQLite/Migrations/M20260905_feeds.hs index 689d3521f0..ef385f6d66 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/M20260905_feeds.hs +++ b/src/Simplex/Chat/Store/SQLite/Migrations/M20260905_feeds.hs @@ -39,40 +39,30 @@ ALTER TABLE groups ADD COLUMN drop_feed INTEGER NOT NULL DEFAULT 0; CREATE INDEX idx_contacts_user_id ON contacts(user_id, contact_id); CREATE INDEX idx_groups_user_id_business_chat ON groups(user_id, business_chat, group_id); -PRAGMA writable_schema=1; - -UPDATE sqlite_master -SET sql = replace(sql, 'group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE', 'group_id INTEGER REFERENCES groups ON DELETE CASCADE') -WHERE type = 'table' AND name = 'delivery_jobs'; - -PRAGMA writable_schema=RESET; - -ALTER TABLE delivery_jobs ADD COLUMN feed_id INTEGER REFERENCES feeds ON DELETE CASCADE; -ALTER TABLE delivery_jobs ADD COLUMN chat_item_id INTEGER REFERENCES chat_items ON DELETE CASCADE; -ALTER TABLE delivery_jobs ADD COLUMN message_ids TEXT; -ALTER TABLE delivery_jobs ADD COLUMN feed_cursor_id INTEGER; -CREATE INDEX idx_delivery_jobs_feed_next ON delivery_jobs(feed_id, worker_scope, failed, job_status); -CREATE INDEX idx_delivery_jobs_chat_item_id ON delivery_jobs(chat_item_id); +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); |] down_m20260905_feeds :: Query down_m20260905_feeds = [sql| -DROP INDEX idx_delivery_jobs_chat_item_id; -DROP INDEX idx_delivery_jobs_feed_next; -DELETE FROM delivery_jobs WHERE group_id IS NULL; -ALTER TABLE delivery_jobs DROP COLUMN feed_cursor_id; -ALTER TABLE delivery_jobs DROP COLUMN message_ids; -ALTER TABLE delivery_jobs DROP COLUMN chat_item_id; -ALTER TABLE delivery_jobs DROP COLUMN feed_id; - -PRAGMA writable_schema=1; - -UPDATE sqlite_master -SET sql = replace(sql, 'group_id INTEGER REFERENCES groups ON DELETE CASCADE', 'group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE') -WHERE type = 'table' AND name = 'delivery_jobs'; - -PRAGMA writable_schema=RESET; +DROP INDEX idx_feed_jobs_chat_item_id; +DROP INDEX idx_feed_jobs_next; +DROP TABLE feed_jobs; DROP INDEX idx_groups_user_id_business_chat; DROP INDEX idx_contacts_user_id; diff --git a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql index 6234fd6c1d..8b918fccfa 100644 --- a/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql +++ b/src/Simplex/Chat/Store/SQLite/Migrations/chat_schema.sql @@ -792,7 +792,7 @@ CREATE TABLE delivery_tasks( ) STRICT; CREATE TABLE delivery_jobs( delivery_job_id INTEGER PRIMARY KEY AUTOINCREMENT, - group_id INTEGER REFERENCES groups ON DELETE CASCADE, + group_id INTEGER NOT NULL REFERENCES groups ON DELETE CASCADE, worker_scope TEXT NOT NULL, job_scope_spec_tag TEXT, job_scope_include_pending INTEGER, @@ -805,11 +805,7 @@ CREATE TABLE delivery_jobs( created_at TEXT NOT NULL DEFAULT(datetime('now')), updated_at TEXT NOT NULL DEFAULT(datetime('now')) , - sender_group_member_ids TEXT, - feed_id INTEGER REFERENCES feeds ON DELETE CASCADE, - chat_item_id INTEGER REFERENCES chat_items ON DELETE CASCADE, - message_ids TEXT, - feed_cursor_id INTEGER + sender_group_member_ids TEXT ) STRICT; CREATE TABLE group_member_status_predicates( member_status TEXT NOT NULL PRIMARY KEY, @@ -872,6 +868,20 @@ CREATE TABLE feeds( favorite INTEGER NOT NULL DEFAULT 0, unread_chat INTEGER NOT NULL DEFAULT 0 ) STRICT; +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 contact_profiles_index ON contact_profiles( display_name, full_name @@ -1429,13 +1439,14 @@ CREATE INDEX idx_groups_user_id_business_chat ON groups( business_chat, group_id ); -CREATE INDEX idx_delivery_jobs_feed_next ON delivery_jobs( +CREATE INDEX idx_feed_jobs_next ON feed_jobs( feed_id, worker_scope, failed, - job_status + job_status, + feed_job_id ); -CREATE INDEX idx_delivery_jobs_chat_item_id ON delivery_jobs(chat_item_id); +CREATE INDEX idx_feed_jobs_chat_item_id ON feed_jobs(chat_item_id, action_tag); CREATE TRIGGER on_group_members_insert_update_summary AFTER INSERT ON group_members FOR EACH ROW diff --git a/src/Simplex/Chat/Store/Shared.hs b/src/Simplex/Chat/Store/Shared.hs index e9a81b64d6..f54450b7d4 100644 --- a/src/Simplex/Chat/Store/Shared.hs +++ b/src/Simplex/Chat/Store/Shared.hs @@ -169,6 +169,8 @@ data StoreError | SEDeliveryTaskNotFound {taskId :: Int64} | SEInvalidDeliveryJob {jobId :: Int64} | SEDeliveryJobNotFound {jobId :: Int64} + | SEInvalidFeedJob {feedJobId :: Int64} + | SEFeedJobNotFound {feedJobId :: Int64} | -- | Error when reading work item that suspends worker - do not use! SEWorkItemError {errContext :: String} deriving (Show, Exception) diff --git a/src/Simplex/Chat/View.hs b/src/Simplex/Chat/View.hs index 76c6e68b37..a827e89cf4 100644 --- a/src/Simplex/Chat/View.hs +++ b/src/Simplex/Chat/View.hs @@ -2442,6 +2442,8 @@ uploadingFile status = \case [status <> " uploading " <> fileTransferStr fileId fileName <> " for " <> ttyContact c] AChatItem _ _ (GroupChat g _scopeInfo) ChatItem {file = Just CIFile {fileId, fileName}, chatDir = CIGroupSnd} -> [status <> " uploading " <> fileTransferStr fileId fileName <> " for " <> ttyGroup' g] + AChatItem _ _ (FeedChat _) ChatItem {file = Just CIFile {fileId, fileName}, chatDir = CIFeedSnd} -> + [status <> " uploading " <> fileTransferStr fileId fileName <> " for " <> ttyTo "%"] _ -> [status <> " uploading file"] uploadingFileStandalone :: StyledString -> FileTransferMeta -> [StyledString] @@ -2845,6 +2847,8 @@ viewChatError isCmd logLevel testView = \case | testView -> ["duplicate group message, group id: " <> sShow groupId <> ", message id: " <> sShow sharedMsgId] | otherwise -> [] SEUserNoteFolderNotFound -> ["no notes folder"] + SEUserFeedNotFound -> ["no feed"] + SEFeedNotFound {feedId} -> ["no feed " <> sShow feedId] SEInternalError {message} | testView && message == "referenced group member not found" -> [] e -> ["chat db error: " <> sShow e] diff --git a/tests/ChatTests/Feed.hs b/tests/ChatTests/Feed.hs index 8fadf097b0..eece573487 100644 --- a/tests/ChatTests/Feed.hs +++ b/tests/ChatTests/Feed.hs @@ -6,6 +6,7 @@ module ChatTests.Feed where import ChatClient import ChatTests.DBUtils import ChatTests.Utils +import qualified Data.ByteString.Char8 as B import Simplex.Chat.Controller (ChatConfig (..)) import Test.Hspec hiding (it) @@ -14,6 +15,10 @@ chatFeedTests = do describe "feed" $ do it "broadcast to contacts and customer groups in several buckets" testFeedBuckets it "edit and delete the broadcast in every chat" testFeedEditDelete + it "recipient with dropped feed receives nothing" testFeedDropped + it "instance edited in its chat is detached from the feed" testFeedDetached + it "broadcast a file to all contacts" testFeedFile + it "incognito contact is excluded from the broadcast" testFeedSkipsIncognito feedTestCfg :: ChatConfig feedTestCfg = testCfg {feedBucketSize = 1} @@ -83,6 +88,118 @@ testFeedEditDelete = chatItems biz "@2" 1 `shouldReturn` [(1, "hello again [marked deleted]")] chatItems biz "#1" 1 `shouldReturn` [(1, "hello again [marked deleted]")] +testFeedDropped :: HasCallStack => TestParams -> IO () +testFeedDropped = + testChatCfg3 feedTestCfg businessProfile aliceProfile bobProfile $ + \biz alice bob -> do + createCCFeed biz + connectUsers biz alice + connectUsers biz bob + + bob ##> "/feed drop @biz on" + bob <## "ok" + + biz `send` "/feed hello everyone" + biz <# "% hello everyone" + alice <# "biz> hello everyone" + (bob "! % (hello everyone) hello again" + biz <# "% [edited] hello again" + alice <# "biz> [edited] hello again" + (bob "\\ % hello again" + biz <### [ConsoleString "message being deleted", ConsoleString "message deleted"] + alice <# "biz> [marked deleted] hello again" + (bob TestParams -> IO () +testFeedDetached = + testChatCfg3 feedTestCfg businessProfile aliceProfile bobProfile $ + \biz alice bob -> do + createCCFeed biz + connectUsers biz alice + connectUsers biz bob + + biz `send` "/feed hello everyone" + biz <# "% hello everyone" + alice <# "biz> hello everyone" + bob <# "biz> hello everyone" + + biz ##> "! @alice (hello everyone) hello alice" + biz <# "@alice [edited] hello alice" + alice <# "biz> [edited] hello alice" + + biz ##> "! % (hello everyone) hello again" + biz <# "% [edited] hello again" + bob <# "biz> [edited] hello again" + (alice TestParams -> IO () +testFeedFile = + testChatCfg3 feedTestCfg businessProfile aliceProfile bobProfile $ + \biz alice bob -> withXFTPServer $ do + createCCFeed biz + connectUsers biz alice + connectUsers biz bob + + biz #> "/f % ./tests/fixtures/test.pdf" + biz <## "use /fc 1 to cancel sending" + alice <# "biz> sends file test.pdf (266.0 KiB / 272376 bytes)" + alice <## "use /fr 1 [/ | ] to receive it" + bob <# "biz> sends file test.pdf (266.0 KiB / 272376 bytes)" + bob <## "use /fr 1 [/ | ] to receive it" + biz <## "completed uploading file 1 (test.pdf) for %" + + alice ##> "/fr 1 ./tests/tmp" + alice + <### [ "saving file 1 from biz to ./tests/tmp/test.pdf", + "started receiving file 1 (test.pdf) from biz" + ] + alice <## "completed receiving file 1 (test.pdf) from biz" + + bob ##> "/fr 1 ./tests/tmp/bob_test.pdf" + bob + <### [ "saving file 1 from biz to ./tests/tmp/bob_test.pdf", + "started receiving file 1 (test.pdf) from biz" + ] + bob <## "completed receiving file 1 (test.pdf) from biz" + + src <- B.readFile "./tests/fixtures/test.pdf" + B.readFile "./tests/tmp/test.pdf" `shouldReturn` src + B.readFile "./tests/tmp/bob_test.pdf" `shouldReturn` src + +testFeedSkipsIncognito :: HasCallStack => TestParams -> IO () +testFeedSkipsIncognito = + testChatCfg3 feedTestCfg businessProfile aliceProfile bobProfile $ + \biz alice bob -> do + createCCFeed biz + connectUsers biz alice + + biz ##> "/c i" + inv <- getInvitation biz + bob ##> ("/c " <> inv) + bob <## "confirmation sent!" + bizIncognito <- getTermLine biz + concurrentlyN_ + [ bob <## (bizIncognito <> ": contact is connected"), + do + biz <## ("bob (Bob): contact is connected, your incognito profile for this contact is " <> bizIncognito) + biz <## "use /i bob to print out this incognito profile again" + ] + + biz `send` "/feed hello everyone" + biz <# "% hello everyone" + alice <# "biz> hello everyone" + (bob TestCC -> String -> Int -> IO [(Int, String)] chatItems cc chatRef count = do cc ##> ("/_get chat " <> chatRef <> " count=" <> show count) diff --git a/tests/ChatTests/Groups.hs b/tests/ChatTests/Groups.hs index 6a721496d0..65504ebd5c 100644 --- a/tests/ChatTests/Groups.hs +++ b/tests/ChatTests/Groups.hs @@ -13052,7 +13052,7 @@ testChannelMemberDeleteEnforcement ps = connId <- relayConnIdToMember bob "dan" ts <- getCurrentTime let ChatController {smpAgent = bobAgent} = chatController bob - chatMsg = ChatMessage chatInitialVRange Nothing (XMsgDel sharedId Nothing Nothing False) + chatMsg = ChatMessage chatInitialVRange Nothing (XMsgDel sharedId Nothing Nothing False Nothing) fwd = GrpMsgForward (FwdMember cathMemId "cath") ts body = encodeBinaryBatch [encodeFwdElement fwd (VMUnsigned chatMsg)] sent <- runExceptT $ sendMessages bobAgent [(connId, PQEncOff, MsgFlags False, vrValue body)] diff --git a/tests/ProtocolTests.hs b/tests/ProtocolTests.hs index 5504eb5d0a..5818f6aa56 100644 --- a/tests/ProtocolTests.hs +++ b/tests/ProtocolTests.hs @@ -258,7 +258,10 @@ decodeChatMessageTest = describe "Chat message encoding/decoding" $ do #==# 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 + #==# XMsgDel (SharedMsgId "\1\2\3\4") Nothing Nothing False Nothing + it "x.msg.del broadcast" $ + "{\"v\":\"9\",\"event\":\"x.msg.del\",\"params\":{\"msgId\":\"AQIDBA==\",\"feed\":true}}" + #==# XMsgDel (SharedMsgId "\1\2\3\4") Nothing Nothing False (Just True) it "x.msg.deleted" $ "{\"v\":\"9\",\"event\":\"x.msg.deleted\",\"params\":{}}" #==# XMsgDeleted