refactor

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

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

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

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

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)