diff --git a/src/Simplex/Chat/Delivery.hs b/src/Simplex/Chat/Delivery.hs index 5dc9bd9199..f94be74894 100644 --- a/src/Simplex/Chat/Delivery.hs +++ b/src/Simplex/Chat/Delivery.hs @@ -271,20 +271,6 @@ feedActionRemovesItem = \case FJADeleteInternal -> True _ -> False -data FeedInstanceSpec - = FISLinked - | FISAny - | FISUndeleted - -feedActionInstances :: FeedJobAction -> FeedInstanceSpec -feedActionInstances = \case - FJANew _ -> FISAny - FJAFileDescr _ -> FISUndeleted - FJAUpdate _ -> FISLinked - FJADeleteBroadcast _ -> FISAny - FJADeleteInternal -> FISAny - FJADeleteMark -> FISAny - instance FromField FeedJobActionTag where fromField = fromTextField_ textDecode instance ToField FeedJobActionTag where toField = toField . textEncode diff --git a/src/Simplex/Chat/Library/Commands.hs b/src/Simplex/Chat/Library/Commands.hs index eee044fbb5..7f1afe1ad0 100644 --- a/src/Simplex/Chat/Library/Commands.hs +++ b/src/Simplex/Chat/Library/Commands.hs @@ -5354,7 +5354,6 @@ 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 @@ -5422,9 +5421,6 @@ 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 diff --git a/src/Simplex/Chat/Library/Subscriber.hs b/src/Simplex/Chat/Library/Subscriber.hs index dbf16bfff2..5436c4e56b 100644 --- a/src/Simplex/Chat/Library/Subscriber.hs +++ b/src/Simplex/Chat/Library/Subscriber.hs @@ -4462,36 +4462,32 @@ sndFeedTimed chatTTL createdAt = feedContactsBucket :: StoreCxt -> User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> Int -> [SndMessage] -> Maybe ContactId -> CM (Maybe ContactId) feedContactsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case action of FJANew _ -> do - cts <- withStore' $ \db -> getFeedContactsByCursor db cxt user cursor_ bucketSize - let ctIds = map contactId' cts - forM_ (lastId ctIds) $ \lastCtId -> do - existing <- withStore' $ \db -> getFeedInstanceContactIdsByRange db user feedItemId fromId lastCtId + cts <- withStore' $ \db -> getFeedContactsByCursor db cxt user feedItemId cursor_ bucketSize + runBucket bucketSize (map (contactId' . fst) cts) $ \lastCtId -> do createdAt <- liftIO getCurrentTime - (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db createdAt existing) (mapMaybe feedRecipient cts)) + (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db createdAt) (mapMaybe feedCtRecipient cts)) unless (null errs) $ toView $ CEvtChatErrors errs forM_ recipients $ \(ctId, itemId, _, newTimed_) -> forM_ (newTimed_ >>= timedDeleteAt') $ startProximateTimedItemThread user (ChatRef CTDirect ctId Nothing, itemId) results <- deliverBucket lastCtId [(ctId, itemId, conn) | (ctId, itemId, conn, _) <- recipients] withStore' $ \db -> updateFeedInstanceStatuses db (sendErrorStatuses results) - pure $ bucketCursor bucketSize ctIds _ -> do - instances <- withStore' $ \db -> getFeedContactInstancesByCursor db cxt user feedItemId (feedActionInstances action) cursor_ bucketSize - let ctIds = map (contactId' . fst) instances - unless (null msgs) $ - forM_ (lastId ctIds) $ \lastCtId -> - void $ deliverBucket lastCtId [(contactId' ct, itemId, conn) | (ct, itemId) <- instances, Just (_, conn) <- [feedRecipient ct]] - applyContactAction user feedItemId action item instances fromId (fromMaybe 0 $ lastId ctIds) - pure $ bucketCursor bucketSize ctIds + instances <- withStore' $ \db -> getFeedContactInstancesByCursor db cxt user feedItemId action cursor_ bucketSize + runBucket bucketSize (map (contactId' . fst) instances) $ \lastCtId -> do + unless (null msgs) $ + void $ deliverBucket lastCtId [(contactId' ct, itemId, conn) | (ct, itemId) <- instances, Just conn <- [feedConn ct]] + applyContactAction user feedItemId action item instances fromId lastCtId where FeedItemMsg {feedSharedMsgId, feedContent, feedHasLink} = item fromId = fromMaybe 0 cursor_ - feedRecipient ct = case contactSendConn_ ct of - Right conn | directOrUsed ct && not (connIncognito conn) -> Just (ct, conn) + feedConn ct = case contactSendConn_ ct of + Right conn | directOrUsed ct && not (connIncognito conn) -> Just conn _ -> Nothing - instanceRecipient db createdAt existing (ct, conn) = + feedCtRecipient (ct, itemId_) = (ct,itemId_,) <$> feedConn ct + instanceRecipient db createdAt (ct, itemId_, conn) = let timed_ = sndFeedTimed (contactTimedTTL ct) createdAt - in case M.lookup (contactId' ct) existing of + in case itemId_ of Just itemId -> pure $ Right (contactId' ct, itemId, conn, Nothing) Nothing -> do itemId <- createFeedInstanceItem db user (CDDirectSnd ct) feedSharedMsgId feedContent feedItemId timed_ feedHasLink createdAt @@ -4506,9 +4502,9 @@ applyContactAction user feedItemId action item instances fromId toId = case acti FJADeleteBroadcast _ -> do let (toDelete, toMark) = partition (fullDelete . fst) instances deleteItems toDelete - markDeleted toMark + markFeedInstances toMark FJADeleteInternal -> deleteItems instances - FJADeleteMark -> markDeleted instances + FJADeleteMark -> markFeedInstances instances _ -> pure () where FeedItemMsg {feedSharedMsgId} = item @@ -4516,20 +4512,15 @@ applyContactAction user feedItemId action item instances fromId toId = case acti deleteItems cts = withStore' $ \db -> do deleteFeedContactReactions db feedSharedMsgId (map (contactId' . fst) cts) deleteFeedInstances db (map snd cts) - markDeleted cts = do - deletedTs <- liftIO getCurrentTime - withStore' $ \db -> markFeedInstancesDeleted db (map snd cts) deletedTs feedGroupsBucket :: StoreCxt -> User -> ChatItemId -> FeedJobAction -> FeedItemMsg -> Int -> [SndMessage] -> Maybe GroupId -> CM (Maybe GroupId) feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case action of FJANew _ -> do - gs <- withStore' $ \db -> getFeedCustomerGroupsByCursor db cxt user cursor_ bucketSize - let gIds = map groupId' gs - forM_ (lastId gIds) $ \lastGId -> do + gs <- withStore' $ \db -> getFeedCustomerGroupsByCursor db cxt user feedItemId cursor_ bucketSize + runBucket bucketSize (map (groupId' . fst) gs) $ \lastGId -> do members <- withStore' $ \db -> getCustomerGroupsMembersByRange db cxt user fromId lastGId - existing <- withStore' $ \db -> getFeedInstanceGroupIdsByRange db user feedItemId fromId lastGId createdAt <- liftIO getCurrentTime - (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db members createdAt existing) (filter feedGroup gs)) + (errs, recipients) <- lift $ partitionEithers <$> withStoreBatch (\db -> map (instanceRecipient db members createdAt) (filter (feedGroup . fst) gs)) unless (null errs) $ toView $ CEvtChatErrors errs forM_ recipients $ \(gId, itemId, _, newTimed_) -> forM_ (newTimed_ >>= timedDeleteAt') $ @@ -4537,16 +4528,13 @@ feedGroupsBucket cxt user feedItemId action item bucketSize msgs cursor_ = case results <- deliverBucket lastGId [(itemId, ms) | (_, itemId, ms, _) <- recipients] 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 - let gIds = map (groupId' . fst) instances - unless (null msgs) $ - forM_ (lastId gIds) $ \lastGId -> do + instances <- withStore' $ \db -> getFeedGroupInstancesByCursor db cxt user feedItemId action cursor_ bucketSize + runBucket bucketSize (map (groupId' . fst) instances) $ \lastGId -> do + unless (null msgs) $ do members <- withStore' $ \db -> getCustomerGroupsMembersByRange db cxt user fromId lastGId void $ deliverBucket lastGId [(itemId, groupMembers members g) | (g, itemId) <- instances] - applyGroupAction user feedItemId action item instances fromId (fromMaybe 0 $ lastId gIds) - pure $ bucketCursor bucketSize gIds + applyGroupAction user feedItemId action item instances fromId lastGId where FeedItemMsg {feedSharedMsgId, feedContent, feedHasLink} = item fromId = fromMaybe 0 cursor_ @@ -4554,10 +4542,10 @@ 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 = + instanceRecipient db members createdAt (g, itemId_) = let ms = groupMembers members g timed_ = sndFeedTimed (groupTimedTTL g) createdAt - in case M.lookup (groupId' g) existing of + in case itemId_ 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 @@ -4574,20 +4562,22 @@ applyGroupAction user feedItemId action item instances fromId toId = case action FJADeleteBroadcast _ -> do let (toDelete, toMark) = partition (groupFeatureUserAllowed SGFFullDelete . fst) instances deleteItems toDelete - markDeleted toMark + markFeedInstances toMark FJADeleteInternal -> deleteItems instances - FJADeleteMark -> markDeleted instances + FJADeleteMark -> markFeedInstances instances _ -> pure () where FeedItemMsg {feedSharedMsgId} = item deleteItems gs = withStore' $ \db -> do deleteFeedGroupReactions db feedSharedMsgId (map (membershipId . fst) gs) deleteFeedInstances db (map snd gs) - markDeleted gs = do - deletedTs <- liftIO getCurrentTime - withStore' $ \db -> markFeedInstancesDeleted db (map snd gs) deletedTs membershipId GroupInfo {groupId, membership} = (groupId, memberId' membership) +markFeedInstances :: [(c, ChatItemId)] -> CM () +markFeedInstances instances = do + deletedTs <- liftIO getCurrentTime + withStore' $ \db -> markFeedInstancesDeleted db (map snd instances) deletedTs + deliverFeedBucket :: FeedJobAction -> [SndMessage] -> [(r, Connection)] -> CM [(r, Either ChatError ([Int64], PQEncryption))] deliverFeedBucket action msgs recipients = case (L.nonEmpty msgs, recipients) of (Just msgs', _ : _) -> do @@ -4617,15 +4607,14 @@ firstMsgId = \case SndMessage {msgId} : _ -> msgId [] -> 0 -lastId :: [Int64] -> Maybe Int64 -lastId ids = case reverse ids of - i : _ -> Just i - [] -> Nothing - -bucketCursor :: Int -> [Int64] -> Maybe Int64 -bucketCursor bucketSize ids - | length ids < bucketSize = Nothing - | otherwise = lastId ids +runBucket :: Int -> [Int64] -> (Int64 -> CM ()) -> CM (Maybe Int64) +runBucket bucketSize ids action = do + forM_ lastId action + pure $ if length ids < bucketSize then Nothing else lastId + where + lastId = case reverse ids of + i : _ -> Just i + [] -> Nothing finishFeedEvent :: User -> Feed -> ChatItemId -> FeedJobAction -> CM () finishFeedEvent user feed@Feed {feedId} feedItemId action diff --git a/src/Simplex/Chat/Store/Feeds.hs b/src/Simplex/Chat/Store/Feeds.hs index 4b42e38aab..03d1cdb7cc 100644 --- a/src/Simplex/Chat/Store/Feeds.hs +++ b/src/Simplex/Chat/Store/Feeds.hs @@ -21,8 +21,6 @@ module Simplex.Chat.Store.Feeds getCustomerGroupsMembersByRange, getFeedContactInstancesByCursor, getFeedGroupInstancesByCursor, - getFeedInstanceContactIdsByRange, - getFeedInstanceGroupIdsByRange, updateFeedInstanceStatuses, getDeliveredContactIdsByRange, getDeliveredMemberIdsByRange, @@ -109,23 +107,37 @@ deleteFeedCIs db User {userId} Feed {feedId} = do DB.execute db "DELETE FROM messages WHERE feed_id = ?" (Only feedId) DB.execute db "DELETE FROM chat_items WHERE user_id = ? AND feed_id = ?" (userId, feedId) -getFeedContactsByCursor :: DB.Connection -> StoreCxt -> User -> Maybe ContactId -> Int -> IO [Contact] -getFeedContactsByCursor db cxt user@User {userId} cursorId_ count = do +getFeedContactsByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> Maybe ContactId -> Int -> IO [(Contact, Maybe ChatItemId)] +getFeedContactsByCursor db cxt user@User {userId} feedItemId cursorId_ count = do currentTs <- getCurrentTime - map (toContact currentTs cxt user []) + map (\(Only itemId_ :. row) -> (toContact currentTs cxt user [] row, itemId_)) <$> DB.query db - (contactQuery <> " WHERE ct.user_id = ? AND ct.deleted = 0 AND ct.is_user = 0 AND ct.contact_id > ? ORDER BY ct.contact_id ASC LIMIT ?") - (userId, cursorId cursorId_, count) + ( "SELECT i.chat_item_id, " + <> contactQueryFields + <> " " + <> contactQueryFrom + <> " LEFT JOIN chat_items i ON i.feed_item_id = ? AND i.contact_id = ct.contact_id" + <> " WHERE ct.user_id = ? AND ct.deleted = 0 AND ct.is_user = 0 AND ct.contact_id > ?" + <> " ORDER BY ct.contact_id ASC LIMIT ?" + ) + (feedItemId, userId, cursorId cursorId_, count) -getFeedCustomerGroupsByCursor :: DB.Connection -> StoreCxt -> User -> Maybe GroupId -> Int -> IO [GroupInfo] -getFeedCustomerGroupsByCursor db cxt User {userId, userContactId} cursorId_ count = do +getFeedCustomerGroupsByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> Maybe GroupId -> Int -> IO [(GroupInfo, Maybe ChatItemId)] +getFeedCustomerGroupsByCursor db cxt User {userId, userContactId} feedItemId cursorId_ count = do currentTs <- getCurrentTime - map (toGroupInfo currentTs cxt userContactId []) + map (\(Only itemId_ :. row) -> (toGroupInfo currentTs cxt userContactId [] row, itemId_)) <$> DB.query db - (groupInfoQuery <> " WHERE g.user_id = ? AND mu.contact_id = ? AND g.business_chat = ? AND g.group_id > ? ORDER BY g.group_id ASC LIMIT ?") - (userId, userContactId, BCCustomer, cursorId cursorId_, count) + ( "SELECT i.chat_item_id, " + <> groupInfoQueryFields + <> " " + <> groupInfoQueryFrom + <> " LEFT JOIN chat_items i ON i.feed_item_id = ? AND i.group_id = g.group_id" + <> " WHERE g.user_id = ? AND mu.contact_id = ? AND g.business_chat = ? AND g.group_id > ?" + <> " ORDER BY g.group_id ASC LIMIT ?" + ) + (feedItemId, userId, userContactId, BCCustomer, cursorId cursorId_, count) getCustomerGroupsMembersByRange :: DB.Connection -> StoreCxt -> User -> GroupId -> GroupId -> IO (Map GroupId [GroupMember]) getCustomerGroupsMembersByRange db cxt user@User {userId, userContactId} fromId toId = do @@ -160,13 +172,13 @@ feedItemMsg (CChatItem _ ChatItem {content, meta = CIMeta {itemSharedMsgId, item <$> itemSharedMsgId _ -> Nothing -instanceSpecCond :: FeedInstanceSpec -> Query +instanceSpecCond :: FeedJobAction -> Query instanceSpecCond = \case - FISLinked -> " AND i.item_feed = 1" - FISAny -> " AND i.item_feed > 0" - FISUndeleted -> " AND i.item_deleted = 0" + FJAUpdate _ -> " AND i.item_feed = 1" + FJAFileDescr _ -> " AND i.item_deleted = 0" + _ -> " AND i.item_feed > 0" -getFeedContactInstancesByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> FeedInstanceSpec -> Maybe ContactId -> Int -> IO [(Contact, ChatItemId)] +getFeedContactInstancesByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> FeedJobAction -> Maybe ContactId -> Int -> IO [(Contact, ChatItemId)] getFeedContactInstancesByCursor db cxt user@User {userId} feedItemId spec cursorId_ count = do currentTs <- getCurrentTime map (\(Only itemId :. row) -> (toContact currentTs cxt user [] row, itemId)) @@ -183,7 +195,7 @@ getFeedContactInstancesByCursor db cxt user@User {userId} feedItemId spec cursor ) (userId, feedItemId, cursorId cursorId_, count) -getFeedGroupInstancesByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> FeedInstanceSpec -> Maybe GroupId -> Int -> IO [(GroupInfo, ChatItemId)] +getFeedGroupInstancesByCursor :: DB.Connection -> StoreCxt -> User -> ChatItemId -> FeedJobAction -> Maybe GroupId -> Int -> IO [(GroupInfo, ChatItemId)] getFeedGroupInstancesByCursor db cxt User {userId, userContactId} feedItemId spec cursorId_ count = do currentTs <- getCurrentTime map (\(Only itemId :. row) -> (toGroupInfo currentTs cxt userContactId [] row, itemId)) @@ -200,23 +212,6 @@ getFeedGroupInstancesByCursor db cxt User {userId, userContactId} feedItemId spe ) (userId, userContactId, feedItemId, cursorId cursorId_, count) -getFeedInstanceContactIdsByRange :: DB.Connection -> User -> ChatItemId -> ContactId -> ContactId -> IO (Map ContactId ChatItemId) -getFeedInstanceContactIdsByRange db user feedItemId = getFeedInstanceIdsByRange_ db user feedItemId "contact_id" - -getFeedInstanceGroupIdsByRange :: DB.Connection -> User -> ChatItemId -> GroupId -> GroupId -> IO (Map GroupId ChatItemId) -getFeedInstanceGroupIdsByRange db user feedItemId = getFeedInstanceIdsByRange_ db user feedItemId "group_id" - -getFeedInstanceIdsByRange_ :: DB.Connection -> User -> ChatItemId -> Query -> Int64 -> Int64 -> IO (Map Int64 ChatItemId) -getFeedInstanceIdsByRange_ db User {userId} feedItemId chatIdColumn fromId toId = - M.fromList - <$> DB.query - db - ( "SELECT " <> chatIdColumn <> ", chat_item_id FROM chat_items" - <> " WHERE user_id = ? AND feed_item_id = ?" - <> " AND " <> chatIdColumn <> " > ? AND " <> chatIdColumn <> " <= ?" - ) - (userId, feedItemId, fromId, toId) - updateFeedInstanceStatuses :: DB.Connection -> [(ChatItemId, CIStatus 'MDSnd)] -> IO () updateFeedInstanceStatuses db statuses = do currentTs <- getCurrentTime @@ -405,20 +400,19 @@ updateFeedJobCursor db jobId cursorId_ = do (cursorId_, currentTs, jobId) setFeedJobErrStatus :: DB.Connection -> Int64 -> Text -> IO () -setFeedJobErrStatus db jobId errReason = do +setFeedJobErrStatus db jobId errReason = updateFeedJobStatus_ db jobId DJSError (Just errReason) + +updateFeedJobStatus_ :: DB.Connection -> Int64 -> DeliveryJobStatus -> Maybe Text -> IO () +updateFeedJobStatus_ db jobId status 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) + (status, 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) + updateFeedJobStatus_ db jobId DJSComplete Nothing unfinished <- maybeFirstRow fromOnly $ DB.query diff --git a/tests/ChatTests/Feed.hs b/tests/ChatTests/Feed.hs index eece573487..6ab33872c5 100644 --- a/tests/ChatTests/Feed.hs +++ b/tests/ChatTests/Feed.hs @@ -30,10 +30,7 @@ testFeedBuckets = createCCFeed biz connectUsers biz alice connectUsers biz bob - biz ##> "/ad" - cLink <- getContactLink biz True - biz ##> "/auto_accept on business" - biz <## "auto_accept on, business" + cLink <- businessAddress biz connectToBusiness biz cath cLink "cath" "Catherine" connectToBusiness biz dan cLink "dan" "Daniel" @@ -57,10 +54,7 @@ testFeedEditDelete = createCCFeed biz connectUsers biz alice connectUsers biz bob - biz ##> "/ad" - cLink <- getContactLink biz True - biz ##> "/auto_accept on business" - biz <## "auto_accept on, business" + cLink <- businessAddress biz connectToBusiness biz cath cLink "cath" "Catherine" biz `send` "/feed hello everyone" @@ -205,6 +199,14 @@ chatItems cc chatRef count = do cc ##> ("/_get chat " <> chatRef <> " count=" <> show count) chat <$> getTermLine cc +businessAddress :: HasCallStack => TestCC -> IO String +businessAddress biz = do + biz ##> "/ad" + cLink <- getContactLink biz True + biz ##> "/auto_accept on business" + biz <## "auto_accept on, business" + pure cLink + connectToBusiness :: HasCallStack => TestCC -> TestCC -> String -> String -> String -> IO () connectToBusiness biz cc cLink name fullName = do cc ##> ("/c " <> cLink)