mirror of
https://code.sup39.dev/repos/Wqawg
synced 2024-12-27 15:04:50 +09:00
Remember for deliveries in the DB, whether they should sign forwarding
This commit is contained in:
fr33domlover
parent
93cf861ed0
commit
5d5c56695e
3 changed files with 50 additions and 34 deletions
|
|
@ -65,16 +65,18 @@ InboxItemRemote
|
|||
UniqueInboxItemRemote person activity
|
||||
|
||||
UnlinkedDelivery
|
||||
recipient UnfetchedRemoteActorId
|
||||
activity OutboxItemId
|
||||
running Bool
|
||||
recipient UnfetchedRemoteActorId
|
||||
activity OutboxItemId
|
||||
forwarding Bool
|
||||
running Bool
|
||||
|
||||
UniqueUnlinkedDelivery recipient activity
|
||||
|
||||
Delivery
|
||||
recipient RemoteActorId
|
||||
activity OutboxItemId
|
||||
running Bool
|
||||
recipient RemoteActorId
|
||||
activity OutboxItemId
|
||||
forwarding Bool
|
||||
running Bool
|
||||
|
||||
UniqueDelivery recipient activity
|
||||
|
||||
|
|
|
|||
|
|
@ -929,7 +929,7 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
moreRemotes <- deliverLocal pid obid localRecips mcollections
|
||||
unless (federation || null moreRemotes) $
|
||||
throwE "Federation disabled but remote collection members found"
|
||||
remotesHttp <- lift $ deliverRemoteDB obid remoteRecips moreRemotes
|
||||
remotesHttp <- lift $ deliverRemoteDB (furiHost uContext) obid remoteRecips moreRemotes
|
||||
return (lmid, obid, doc, remotesHttp)
|
||||
(lmid, obid, doc, remotesHttp) <- case result of
|
||||
Left (FedError t) -> throwE t
|
||||
|
|
@ -1318,7 +1318,8 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
Right _gid -> throwE "Local Note addresses a local group"
|
||||
|
||||
deliverRemoteDB
|
||||
:: OutboxItemId
|
||||
:: Text
|
||||
-> OutboxItemId
|
||||
-> [FedURI]
|
||||
-> [((InstanceId, Text), NonEmpty (RemoteActorId, LocalURI, LocalURI, Maybe UTCTime))]
|
||||
-> AppDB
|
||||
|
|
@ -1326,7 +1327,7 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
, [((InstanceId, Text), NonEmpty (UnfetchedRemoteActorId, LocalURI, UnlinkedDeliveryId))]
|
||||
, [((InstanceId, Text), NonEmpty (UnfetchedRemoteActorId, LocalURI, UnlinkedDeliveryId))]
|
||||
)
|
||||
deliverRemoteDB obid recips known = do
|
||||
deliverRemoteDB hContext obid recips known = do
|
||||
recips' <- for (groupByHost recips) $ \ (h, lus) -> do
|
||||
let lus' = NE.nub lus
|
||||
(iid, inew) <- idAndNew <$> insertBy' (Instance h)
|
||||
|
|
@ -1353,13 +1354,16 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
-- TODO see the earlier TODO about merge, it applies here too
|
||||
allFetched = map (second $ NE.nubBy ((==) `on` fst4)) $ mergeConcat known moreKnown
|
||||
fetchedDeliv <- for allFetched $ \ (i, rs) ->
|
||||
(i,) <$> insertMany' (\ (raid, _, _, msince) -> Delivery raid obid $ isNothing msince) rs
|
||||
let fwd = snd i == hContext
|
||||
in (i,) <$> insertMany' (\ (raid, _, _, msince) -> Delivery raid obid fwd $ isNothing msince) rs
|
||||
unfetchedDeliv <- for unfetched $ \ (i, rs) ->
|
||||
(i,) <$> insertMany' (\ (uraid, _, msince) -> UnlinkedDelivery uraid obid $ isNothing msince) rs
|
||||
let fwd = snd i == hContext
|
||||
in (i,) <$> insertMany' (\ (uraid, _, msince) -> UnlinkedDelivery uraid obid fwd $ isNothing msince) rs
|
||||
unknownDeliv <- for stillUnknown $ \ (i, lus) -> do
|
||||
-- TODO maybe for URA insertion we should do insertUnique?
|
||||
rs <- insertMany' (\ lu -> UnfetchedRemoteActor (fst i) lu Nothing) lus
|
||||
(i,) <$> insertMany' (\ (_, uraid) -> UnlinkedDelivery uraid obid True) rs
|
||||
let fwd = snd i == hContext
|
||||
(i,) <$> insertMany' (\ (_, uraid) -> UnlinkedDelivery uraid obid fwd True) rs
|
||||
return
|
||||
( takeNoError4 fetchedDeliv
|
||||
, takeNoError3 unfetchedDeliv
|
||||
|
|
@ -1398,8 +1402,9 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
)
|
||||
-> Handler ()
|
||||
deliverRemoteHttp hContext obid doc (fetched, unfetched, unknown) = do
|
||||
let deliver fwd h =
|
||||
deliverHttp doc (if h == hContext then Just fwd else Nothing) h
|
||||
let deliver fwd h inbox = do
|
||||
let fwd' = if h == hContext then Just fwd else Nothing
|
||||
(isJust fwd',) <$> deliverHttp doc fwd' h inbox
|
||||
now <- liftIO getCurrentTime
|
||||
traverse_ (fork . deliverFetched deliver now) fetched
|
||||
traverse_ (fork . deliverUnfetched deliver now) unfetched
|
||||
|
|
@ -1408,7 +1413,7 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
fork = forkHandler $ \ e -> logError $ "Outbox POST handler: delivery failed! " <> T.pack (displayException e)
|
||||
deliverFetched deliver now ((_, h), recips@(r :| rs)) = do
|
||||
let (raid, luActor, luInbox, dlid) = r
|
||||
e <- deliver luActor h luInbox
|
||||
(_, e) <- deliver luActor h luInbox
|
||||
let e' = case e of
|
||||
Left err ->
|
||||
if isInstanceErrorP err
|
||||
|
|
@ -1429,7 +1434,7 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
update dlid [DeliveryRunning =. False]
|
||||
for_ rs $ \ (raid, luActor, luInbox, dlid) ->
|
||||
fork $ do
|
||||
e <- deliver luActor h luInbox
|
||||
(_, e) <- deliver luActor h luInbox
|
||||
runDB $
|
||||
case e of
|
||||
Left _err -> do
|
||||
|
|
@ -1457,13 +1462,13 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
e <- fetchRemoteActor iid h luActor
|
||||
case e of
|
||||
Right (Right (Entity raid ra)) -> do
|
||||
e' <- deliver luActor h $ remoteActorInbox ra
|
||||
(fwd, e') <- deliver luActor h $ remoteActorInbox ra
|
||||
runDB $
|
||||
case e' of
|
||||
Left _ -> do
|
||||
updateWhere [RemoteActorId ==. raid, RemoteActorErrorSince ==. Nothing] [RemoteActorErrorSince =. Just now]
|
||||
delete udlid
|
||||
insert_ $ Delivery raid obid False
|
||||
insert_ $ Delivery raid obid fwd False
|
||||
Right _ -> delete udlid
|
||||
_ -> runDB $ do
|
||||
updateWhere [UnfetchedRemoteActorId ==. uraid, UnfetchedRemoteActorSince ==. Nothing] [UnfetchedRemoteActorSince =. Just now]
|
||||
|
|
@ -1473,13 +1478,13 @@ handleOutboxNote host (Note mluNote luAttrib aud muParent muContext mpublished c
|
|||
updateWhere [UnfetchedRemoteActorId ==. uraid, UnfetchedRemoteActorSince ==. Nothing] [UnfetchedRemoteActorSince =. Just now]
|
||||
update udlid [UnlinkedDeliveryRunning =. False]
|
||||
Just (Entity raid ra) -> do
|
||||
e'' <- deliver luActor h $ remoteActorInbox ra
|
||||
(fwd, e'') <- deliver luActor h $ remoteActorInbox ra
|
||||
runDB $
|
||||
case e'' of
|
||||
Left _ -> do
|
||||
updateWhere [RemoteActorId ==. raid, RemoteActorErrorSince ==. Nothing] [RemoteActorErrorSince =. Just now]
|
||||
delete udlid
|
||||
insert_ $ Delivery raid obid False
|
||||
insert_ $ Delivery raid obid fwd False
|
||||
Right _ -> delete udlid
|
||||
|
||||
retryOutboxDelivery :: Worker ()
|
||||
|
|
@ -1504,6 +1509,7 @@ retryOutboxDelivery = do
|
|||
, ura E.^. UnfetchedRemoteActorSince
|
||||
, udl E.^. UnlinkedDeliveryId
|
||||
, udl E.^. UnlinkedDeliveryActivity
|
||||
, udl E.^. UnlinkedDeliveryForwarding
|
||||
, ob E.^. OutboxItemActivity
|
||||
, ra E.?. RemoteActorId
|
||||
)
|
||||
|
|
@ -1534,9 +1540,11 @@ retryOutboxDelivery = do
|
|||
( i E.^. InstanceId
|
||||
, i E.^. InstanceHost
|
||||
, ra E.^. RemoteActorId
|
||||
, ra E.^. RemoteActorIdent
|
||||
, ra E.^. RemoteActorInbox
|
||||
, ra E.^. RemoteActorErrorSince
|
||||
, dl E.^. DeliveryId
|
||||
, dl E.^. DeliveryForwarding
|
||||
, ob E.^. OutboxItemActivity
|
||||
)
|
||||
let (linkedOld, linkedNew) = partitionEithers $ map (decideBySinceDL dropAfter now . adaptLinked) linked
|
||||
|
|
@ -1551,18 +1559,18 @@ retryOutboxDelivery = do
|
|||
unless (and resultsUDL) $ logError "Periodic delivery UDL error"
|
||||
where
|
||||
adaptUnlinked
|
||||
(E.Value iid, E.Value h, E.Value uraid, E.Value luRecip, E.Value since, E.Value udlid, E.Value obid, E.Value act, E.Value mraid) =
|
||||
(E.Value iid, E.Value h, E.Value uraid, E.Value luRecip, E.Value since, E.Value udlid, E.Value obid, E.Value fwd, E.Value act, E.Value mraid) =
|
||||
( mraid
|
||||
, ( ( (iid, h)
|
||||
, ((uraid, luRecip), (udlid, obid, persistJSONValue act))
|
||||
, ((uraid, luRecip), (udlid, fwd, obid, persistJSONValue act))
|
||||
)
|
||||
, since
|
||||
)
|
||||
)
|
||||
unlinkedID ((_, (_, (udlid, _, _))), _) = udlid
|
||||
toLinked (raid, ((_, (_, (_, obid, _))), _)) = Delivery raid obid False
|
||||
unlinkedID ((_, (_, (udlid, _, _, _))), _) = udlid
|
||||
toLinked (raid, ((_, (_, (_, fwd, obid, _))), _)) = Delivery raid obid fwd False
|
||||
relevant dropAfter now since = addUTCTime dropAfter since > now
|
||||
decideBySinceUDL dropAfter now (udl@(_, (_, (udlid, _, _))), msince) =
|
||||
decideBySinceUDL dropAfter now (udl@(_, (_, (udlid, _, _, _))), msince) =
|
||||
case msince of
|
||||
Nothing -> Right udl
|
||||
Just since ->
|
||||
|
|
@ -1573,13 +1581,13 @@ retryOutboxDelivery = do
|
|||
= map (second $ groupWithExtractBy1 ((==) `on` fst) fst snd)
|
||||
. groupWithExtractBy ((==) `on` fst) fst snd
|
||||
adaptLinked
|
||||
(E.Value iid, E.Value h, E.Value raid, E.Value inbox, E.Value since, E.Value dlid, E.Value act) =
|
||||
(E.Value iid, E.Value h, E.Value raid, E.Value ident, E.Value inbox, E.Value since, E.Value dlid, E.Value fwd, E.Value act) =
|
||||
( ( (iid, h)
|
||||
, ((raid, inbox), (dlid, persistJSONValue act))
|
||||
, ((raid, (ident, inbox)), (dlid, fwd, persistJSONValue act))
|
||||
)
|
||||
, since
|
||||
)
|
||||
decideBySinceDL dropAfter now (dl@(_, (_, (dlid, _))), msince) =
|
||||
decideBySinceDL dropAfter now (dl@(_, (_, (dlid, _, _))), msince) =
|
||||
case msince of
|
||||
Nothing -> Right dl
|
||||
Just since ->
|
||||
|
|
@ -1599,9 +1607,10 @@ retryOutboxDelivery = do
|
|||
return False
|
||||
Right success -> return success
|
||||
deliverLinked deliver now ((_, h), recips) = do
|
||||
waitsR <- for recips $ \ ((raid, inbox), delivs) -> fork $ do
|
||||
waitsD <- for delivs $ \ (dlid, doc) -> fork $ do
|
||||
e <- deliver doc Nothing h inbox
|
||||
waitsR <- for recips $ \ ((raid, (ident, inbox)), delivs) -> fork $ do
|
||||
waitsD <- for delivs $ \ (dlid, fwd, doc) -> fork $ do
|
||||
let fwd' = if fwd then Just ident else Nothing
|
||||
e <- deliver doc fwd' h inbox
|
||||
case e of
|
||||
Left _err -> return False
|
||||
Right _resp -> do
|
||||
|
|
@ -1624,13 +1633,14 @@ retryOutboxDelivery = do
|
|||
e <- fetchRemoteActor iid h luRecip
|
||||
case e of
|
||||
Right (Right (Entity raid ra)) -> do
|
||||
waitsD <- for delivs $ \ (udlid, obid, doc) -> fork $ do
|
||||
e' <- deliver doc Nothing h $ remoteActorInbox ra
|
||||
waitsD <- for delivs $ \ (udlid, fwd, obid, doc) -> fork $ do
|
||||
let fwd' = if fwd then Just luRecip else Nothing
|
||||
e' <- deliver doc fwd' h $ remoteActorInbox ra
|
||||
case e' of
|
||||
Left _err -> do
|
||||
runSiteDB $ do
|
||||
delete udlid
|
||||
insert_ $ Delivery raid obid False
|
||||
insert_ $ Delivery raid obid fwd False
|
||||
return False
|
||||
Right _resp -> do
|
||||
runSiteDB $ delete udlid
|
||||
|
|
|
|||
|
|
@ -253,6 +253,10 @@ changes =
|
|||
, removeField "RemoteMessage" "raw"
|
||||
-- 63
|
||||
, removeEntity "RemoteRawObject"
|
||||
-- 64
|
||||
, addFieldPrimRequired "UnlinkedDelivery" True "forwarding"
|
||||
-- 65
|
||||
, addFieldPrimRequired "Delivery" True "forwarding"
|
||||
]
|
||||
|
||||
migrateDB :: MonadIO m => ReaderT SqlBackend m (Either Text (Int, Int))
|
||||
|
|
|
|||
Loading…
Reference in a new issue