diff --git a/cabal.project b/cabal.project index 1cf1aa8ecbc..fb41d725c8e 100644 --- a/cabal.project +++ b/cabal.project @@ -1,6 +1,7 @@ repository hackage.haskell.org url: https://hackage.haskell.org/ index-state: 2023-10-03T15:17:00Z + packages: integration , libs/bilge/ @@ -96,3 +97,14 @@ package wire-message-proto-lens flags: +nix-dev-env package types-common-journal flags: +nix-dev-env + +source-repository-package + type: git + location: https://github.com/velveteer/arbiter.git + tag: 96097411a0480e182d0cbce819c45023fe8aeec7 + subdir: + arbiter-core + arbiter-hasql + arbiter-migrations + arbiter-worker + arbiter-test-common diff --git a/changelog.d/0-release-notes/WPB-26489 b/changelog.d/0-release-notes/WPB-26489 new file mode 100644 index 00000000000..f03cf536a19 --- /dev/null +++ b/changelog.d/0-release-notes/WPB-26489 @@ -0,0 +1 @@ +Background-worker scheduled jobs now run through Arbiter. The background-worker configuration includes `scheduledJobs.pollInterval`, which controls how often the scheduler wakes up to look for due jobs. diff --git a/changelog.d/2-features/WPB-26489 b/changelog.d/2-features/WPB-26489 new file mode 100644 index 00000000000..14def7ff4c3 --- /dev/null +++ b/changelog.d/2-features/WPB-26489 @@ -0,0 +1 @@ +Introduce Arbiter-backed schedulable background jobs, migrate meetings cleanup to the new job runner, and add the initial adminless reminder and deletion jobs. diff --git a/charts/wire-server/templates/background-worker/configmap.yaml b/charts/wire-server/templates/background-worker/configmap.yaml index e7e6f1d2262..ae4bee7d2e0 100644 --- a/charts/wire-server/templates/background-worker/configmap.yaml +++ b/charts/wire-server/templates/background-worker/configmap.yaml @@ -93,6 +93,10 @@ data: backgroundJobs: {{ toYaml . | indent 6 }} {{- end }} + scheduledJobs: + # Arbiter dispatcher poll interval for all scheduled jobs. + # Lower values reduce discovery latency, but increase DB polling. + pollInterval: {{ .scheduledJobs.pollInterval }} {{- with .meetingsCleanup }} meetingsCleanup: {{ toYaml . | indent 6 }} diff --git a/charts/wire-server/values.yaml b/charts/wire-server/values.yaml index 59274ebf5fc..18e62025816 100644 --- a/charts/wire-server/values.yaml +++ b/charts/wire-server/values.yaml @@ -1002,6 +1002,12 @@ background-worker: # Total attempts, including the first try maxAttempts: 3 + # Scheduled-jobs dispatcher configuration. + scheduledJobs: + # Arbiter dispatcher poll interval for all scheduled jobs. + # Lower values reduce discovery latency, but increase DB polling. + pollInterval: 5s + # Meetings cleanup configuration meetingsCleanup: # Delete meetings older than this many hours (48 hours = 2 days) diff --git a/docs/src/developer/reference/config-options.md b/docs/src/developer/reference/config-options.md index 407e7d9ef4e..30e11bcb597 100644 --- a/docs/src/developer/reference/config-options.md +++ b/docs/src/developer/reference/config-options.md @@ -2235,3 +2235,4 @@ Notes - The `migrate...` flags control the corresponding PostgreSQL backfill jobs for the current migration settings; leave them `false` for new installs and after migration. - `concurrency`, `jobTimeout`, and `maxAttempts` control parallelism and retry behavior of the consumer. - `brig` and `gundeck` endpoints default to in-cluster services; override via `background-worker.config.brig` and `.gundeck` if your service DNS/ports differ. +- `scheduledJobs.pollInterval` controls how often the background worker wakes up to check Arbiter for due jobs. diff --git a/flake.lock b/flake.lock index f4c3b235f89..49d3242d64c 100644 --- a/flake.lock +++ b/flake.lock @@ -16,6 +16,23 @@ "type": "github" } }, + "arbiter": { + "flake": false, + "locked": { + "lastModified": 1782164937, + "narHash": "sha256-6TOIjB+p0+yZo58hNOWshsa7JWTtMM5wYpkdEEcH8Pk=", + "owner": "velveteer", + "repo": "arbiter", + "rev": "96097411a0480e182d0cbce819c45023fe8aeec7", + "type": "github" + }, + "original": { + "owner": "velveteer", + "repo": "arbiter", + "rev": "96097411a0480e182d0cbce819c45023fe8aeec7", + "type": "github" + } + }, "bloodhound": { "flake": false, "locked": { @@ -367,6 +384,7 @@ "root": { "inputs": { "amazonka": "amazonka", + "arbiter": "arbiter", "bloodhound": "bloodhound", "cql": "cql", "cql-io": "cql-io", diff --git a/flake.nix b/flake.nix index 5206ca874e8..94ae0cf0e0f 100644 --- a/flake.nix +++ b/flake.nix @@ -108,6 +108,11 @@ url = "github:wireapp/hasql-resource-pool?rev=5b5d3df0fff81801986a0110acae5420215f01c5"; flake = false; }; + + arbiter = { + url = "github:velveteer/arbiter?rev=96097411a0480e182d0cbce819c45023fe8aeec7"; + flake = false; + }; }; outputs = inputs@{ nixpkgs, nixpkgs_24_11, flake-utils, tom-bombadil, sbomnix, ... }: diff --git a/integration/test/Notifications.hs b/integration/test/Notifications.hs index 67281da2d6e..04db1074594 100644 --- a/integration/test/Notifications.hs +++ b/integration/test/Notifications.hs @@ -205,6 +205,9 @@ isConvCreateNotifNotSelf n = isConvDeleteNotif :: (HasCallStack, MakesValue a) => a -> App Bool isConvDeleteNotif n = fieldEquals n "payload.0.type" "conversation.delete" +isConvAdminlessReminderNotif :: (HasCallStack, MakesValue a) => a -> App Bool +isConvAdminlessReminderNotif n = fieldEquals n "payload.0.type" "conversation.adminless-reminder" + notifTypeIsEqual :: (HasCallStack, MakesValue a) => String -> a -> App Bool notifTypeIsEqual typ n = nPayload n %. "type" `isEqual` typ diff --git a/integration/test/Test/AdminlessGroups.hs b/integration/test/Test/AdminlessGroups.hs index d2bd97e5e21..31cee6a75f7 100644 --- a/integration/test/Test/AdminlessGroups.hs +++ b/integration/test/Test/AdminlessGroups.hs @@ -20,7 +20,9 @@ module Test.AdminlessGroups where import API.Brig import API.Galley import API.GalleyInternal hiding (getConversation) +import qualified API.GalleyInternal as GalleyI import MLS.Util +import Notifications import SetupHelpers hiding (deleteUser) import Testlib.Prelude @@ -100,19 +102,50 @@ testOnLastAdminLeaveNoEligibleMembersExist = do (alice, tid, _) <- createTeam OwnDomain 1 setTeamFeatureLockStatus alice tid "preventAdminlessGroups" "unlocked" - patchTeamFeature OwnDomain tid "preventAdminlessGroups" (object ["status" .= "enabled"]) >>= assertSuccess + patchTeamFeature + OwnDomain + tid + "preventAdminlessGroups" + ( object + [ "status" .= "enabled", + "config" + .= object + -- The reminders are due early (+1s and +2s), while deletion is + -- later (+10s). This gives Arbiter's 1s polling and serial + -- grouped-job processing enough room to emit both reminders + -- before the conversation is deleted. + [ "deletionTimeoutDuration" .= "10s", + "reminderTimeoutDurations" .= ["9s", "8s"], + "promotionStrategy" .= "random" + ] + ] + ) + >>= assertSuccess + + let newApp :: NewApp + newApp = def {name = "adminless-reminder-app", description = "non-eligible reminder recipient"} + app <- bindResponse (createApp alice tid newApp) $ \resp -> do + resp.status `shouldMatchInt` 200 + resp.json %. "user" - alice1 <- createMLSClient def alice - void $ uploadNewKeyPackage def alice1 + [alice1, app1] <- traverse (createMLSClient def) [alice, app] + traverse_ (uploadNewKeyPackage def) [alice1, app1] conv <- postConversation alice defMLS {team = Just tid} >>= getJSON 201 convId <- objConvId conv createGroup def alice1 convId - void $ createAddCommit alice1 convId [] >>= sendAndConsumeCommitBundle + void $ createAddCommit alice1 convId [app] >>= sendAndConsumeCommitBundle - -- alice leaves the conversation, no error, group will be marked for deletion - bindResponse (removeMember alice conv alice) $ \resp -> do - resp.status `shouldMatchInt` 200 + withWebSockets [app] $ \[wsApp] -> do + -- alice leaves the conversation, no error, group will be marked for deletion + bindResponse (removeMember alice conv alice) $ \resp -> do + resp.status `shouldMatchInt` 200 + + void $ awaitNMatches 2 isConvAdminlessReminderNotif wsApp + + retryT $ do + bindResponse (GalleyI.getConversation conv) $ \resp -> do + resp.status `shouldMatchInt` 404 testOnLastAdminLeaveFeatureDisabled :: (HasCallStack) => App () testOnLastAdminLeaveFeatureDisabled = do diff --git a/libs/extended/src/Hasql/Pool/Extended.hs b/libs/extended/src/Hasql/Pool/Extended.hs index 0cd8e4b61c9..31c6bf67cab 100644 --- a/libs/extended/src/Hasql/Pool/Extended.hs +++ b/libs/extended/src/Hasql/Pool/Extended.hs @@ -18,8 +18,9 @@ module Hasql.Pool.Extended where import Data.Aeson -import Data.Map as Map +import Data.Map qualified as Map import Data.Misc +import Data.Secret (SecretText, secretText) import Hasql.Connection qualified import Hasql.Connection.Settings qualified as HasqlConnSettings import Hasql.Pool qualified as HasqlPool @@ -47,6 +48,17 @@ instance FromJSON PoolConfig where <*> o .: "acquisitionTimeout" <*> o .: "idlenessTimeout" +-- | Render a PostgreSQL connection string in libpq key-value format. +-- +-- Passwords from the optional secret file are inserted into the key-value map +-- before rendering. The result is wrapped because it may contain the password. +postgresqlConnectionStringWithPassword :: Map Text Text -> Maybe FilePathSecrets -> IO SecretText +postgresqlConnectionStringWithPassword pgConfig mFpSecrets = do + mPw <- for mFpSecrets initCredentials + let pgConfig' = maybe pgConfig (\pw -> Map.insert "password" pw pgConfig) mPw + pure . secretText . PostgresqlConnectionString.toKeyValueString $ + PostgresqlConnectionString.fromKeyValueParams pgConfig' + data HasqlPoolMetrics = HasqlPoolMetrics { readyForUseGauge :: Gauge, inUseGauge :: Gauge, diff --git a/libs/types-common/src/Data/Secret.hs b/libs/types-common/src/Data/Secret.hs new file mode 100644 index 00000000000..73df5920d22 --- /dev/null +++ b/libs/types-common/src/Data/Secret.hs @@ -0,0 +1,43 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Data.Secret + ( SecretText, + secretText, + revealSecretText, + ) +where + +import Imports + +-- | Text that may contain credentials or other sensitive material. +-- +-- The constructor is intentionally opaque. 'revealSecretText' should only be +-- used at the narrow boundary where an external API requires the plaintext +-- representation. +newtype SecretText = SecretText Text + +instance Show SecretText where + show _ = "" + +-- | Wrap sensitive text without exposing it through the public constructor. +secretText :: Text -> SecretText +secretText = SecretText + +-- | Reveal sensitive text for an API that explicitly requires plaintext. +revealSecretText :: SecretText -> Text +revealSecretText (SecretText value) = value diff --git a/libs/types-common/test/Main.hs b/libs/types-common/test/Main.hs index 4814492dfd0..41bd84e1939 100644 --- a/libs/types-common/test/Main.hs +++ b/libs/types-common/test/Main.hs @@ -23,6 +23,7 @@ where import Imports import Test.Data.Mailbox qualified as Mailbox import Test.Data.PEMKeys qualified as PEMKeys +import Test.Data.Secret qualified as Secret import Test.Domain qualified as Domain import Test.Handle qualified as Handle import Test.Properties qualified as Properties @@ -41,5 +42,6 @@ main = Handle.tests, Qualified.tests, PEMKeys.tests, - Mailbox.tests + Mailbox.tests, + Secret.tests ] diff --git a/libs/types-common/test/Test/Data/Secret.hs b/libs/types-common/test/Test/Data/Secret.hs new file mode 100644 index 00000000000..65ab82c7cdc --- /dev/null +++ b/libs/types-common/test/Test/Data/Secret.hs @@ -0,0 +1,31 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Test.Data.Secret (tests) where + +import Data.Secret +import Imports +import Test.Tasty +import Test.Tasty.HUnit + +tests :: TestTree +tests = + testGroup + "SecretText" + [ testCase "does not expose its value through Show" $ + show (secretText "database-password") @?= "" + ] diff --git a/libs/types-common/types-common.cabal b/libs/types-common/types-common.cabal index e2249067182..7d1590c0571 100644 --- a/libs/types-common/types-common.cabal +++ b/libs/types-common/types-common.cabal @@ -31,6 +31,7 @@ library Data.Qualified Data.Range Data.RetryAfter + Data.Secret Data.SizedHashMap Data.Text.Ascii Data.UUID.Tagged @@ -160,6 +161,7 @@ test-suite types-common-tests Paths_types_common Test.Data.Mailbox Test.Data.PEMKeys + Test.Data.Secret Test.Domain Test.Handle Test.Properties diff --git a/libs/wire-api/src/Wire/API/Event/Conversation.hs b/libs/wire-api/src/Wire/API/Event/Conversation.hs index a75477d2ade..5275806cfd2 100644 --- a/libs/wire-api/src/Wire/API/Event/Conversation.hs +++ b/libs/wire-api/src/Wire/API/Event/Conversation.hs @@ -68,6 +68,7 @@ module Wire.API.Event.Conversation MemberUpdateData (..), OtrMessage (..), ConversationReset (..), + AdminlessReminder (..), -- * re-exports ConversationReceiptModeUpdate (..), @@ -194,6 +195,7 @@ data EventType | ProtocolUpdate | AddPermissionUpdate | ConvHistoryUpdate + | ConvAdminlessReminder deriving stock (Eq, Show, Generic, Enum, Bounded, Ord) deriving (Arbitrary) via (GenericUniform EventType) deriving (FromJSON, ToJSON, S.ToSchema) via Schema EventType @@ -222,7 +224,8 @@ instance ToSchema EventType where element "conversation.mls-welcome" MLSWelcome, element "conversation.protocol-update" ProtocolUpdate, element "conversation.add-permission-update" AddPermissionUpdate, - element "conversation.history-update" ConvHistoryUpdate + element "conversation.history-update" ConvHistoryUpdate, + element "conversation.adminless-reminder" ConvAdminlessReminder ] data EventData @@ -247,6 +250,7 @@ data EventData | EdProtocolUpdate P.ProtocolTag | EdAddPermissionUpdate Conv.AddPermissionUpdate | EdConvHistoryUpdate History + | EdAdminlessReminder AdminlessReminder deriving stock (Eq, Show, Generic) genEventData :: EventType -> QC.Gen EventData @@ -272,6 +276,7 @@ genEventData = \case ProtocolUpdate -> EdProtocolUpdate <$> arbitrary AddPermissionUpdate -> EdAddPermissionUpdate <$> arbitrary ConvHistoryUpdate -> EdConvHistoryUpdate <$> arbitrary + ConvAdminlessReminder -> EdAdminlessReminder <$> arbitrary eventDataType :: EventData -> EventType eventDataType (EdMembersJoin _) = MemberJoin @@ -295,6 +300,7 @@ eventDataType (EdConvReset _) = ConvReset eventDataType (EdProtocolUpdate _) = ProtocolUpdate eventDataType (EdAddPermissionUpdate _) = AddPermissionUpdate eventDataType (EdConvHistoryUpdate _) = ConvHistoryUpdate +eventDataType (EdAdminlessReminder _) = ConvAdminlessReminder createConversationEventData :: OwnConversation GroupConvType -> EventData @@ -326,6 +332,7 @@ isCellsConversationEvent eventType = ProtocolUpdate -> False AddPermissionUpdate -> False ConvHistoryUpdate -> False + ConvAdminlessReminder -> False -------------------------------------------------------------------------------- -- Event data helpers @@ -493,6 +500,19 @@ instance ToSchema ConversationReset where <$> (.groupId) .= field "group_id" schema <*> (.newGroupId) .= maybe_ (optField "new_group_id" schema) +data AdminlessReminder = AdminlessReminder + { deletionScheduledFor :: UTCTimeMillis + } + deriving stock (Eq, Show, Generic) + deriving (Arbitrary) via (GenericUniform AdminlessReminder) + deriving (FromJSON, ToJSON, S.ToSchema) via Schema AdminlessReminder + +instance ToSchema AdminlessReminder where + schema = + object $ + AdminlessReminder + <$> (.deletionScheduledFor) .= field "deletion_scheduled_for" schema + makePrisms ''EventData taggedEventDataSchema :: ObjectSchema SwaggerDoc (EventType, EventData) @@ -529,6 +549,7 @@ taggedEventDataSchema = ProtocolUpdate -> tag _EdProtocolUpdate (unnamed (unProtocolUpdate <$> P.ProtocolUpdate .= schema)) AddPermissionUpdate -> tag _EdAddPermissionUpdate (unnamed schema) ConvHistoryUpdate -> tag _EdConvHistoryUpdate (unnamed schema) + ConvAdminlessReminder -> tag _EdAdminlessReminder (unnamed schema) memberLeaveSchema :: ValueSchema NamedSwaggerDoc (EdMemberLeftReason, QualifiedUserIdList) memberLeaveSchema = diff --git a/libs/wire-api/src/Wire/API/Jobs.hs b/libs/wire-api/src/Wire/API/Jobs.hs new file mode 100644 index 00000000000..2a18c0cefef --- /dev/null +++ b/libs/wire-api/src/Wire/API/Jobs.hs @@ -0,0 +1,111 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE DeriveGeneric #-} +{-# LANGUAGE TypeFamilies #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License +-- for more details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.API.Jobs where + +import Data.Aeson (FromJSON, ToJSON, Value (Null), parseJSON, toJSON) +import Data.Id +import Data.Json.Util +import Data.OpenApi qualified as S +import Data.Schema +import Imports +import Test.QuickCheck (Arbitrary (..)) + +-- | Shared queue name for the scheduled meetings cleanup job. +meetingsCleanupQueueName :: Text +meetingsCleanupQueueName = "meetings_cleanup_jobs" + +-- | Shared queue name for the adminless deletion job. +adminlessDeletionQueueName :: Text +adminlessDeletionQueueName = "adminless_deletion_jobs" + +-- | Shared queue name for the adminless reminder job. +adminlessReminderQueueName :: Text +adminlessReminderQueueName = "adminless_reminder_jobs" + +-- | Empty payload because the schedule itself carries all execution context. +data MeetingsCleanupJob = MeetingsCleanupJob + deriving stock (Eq, Generic, Show) + +instance Arbitrary MeetingsCleanupJob where + arbitrary = pure MeetingsCleanupJob + +instance ToJSON MeetingsCleanupJob where + toJSON MeetingsCleanupJob = Null + +instance FromJSON MeetingsCleanupJob where + parseJSON Null = pure MeetingsCleanupJob + parseJSON _ = fail "MeetingsCleanupJob expects null" + +-- | Payload for adminless deletions. +-- Keep the JSON encoding backwards compatible. Arbiter persists these payloads +-- and workers decode them later, so changing field names or shapes without a +-- migration will break already scheduled jobs. +data AdminlessDeletionJob = AdminlessDeletionJob + { adminlessDeletionJobTeamId :: TeamId, + adminlessDeletionJobConversationId :: ConvId, + adminlessDeletionJobOrigUserId :: UserId + } + deriving stock (Eq, Generic, Show) + deriving (ToJSON, FromJSON, S.ToSchema) via (Schema AdminlessDeletionJob) + +instance Arbitrary AdminlessDeletionJob where + arbitrary = AdminlessDeletionJob <$> arbitrary <*> arbitrary <*> arbitrary + +instance ToSchema AdminlessDeletionJob where + schema = + object $ + AdminlessDeletionJob + <$> (.adminlessDeletionJobTeamId) .= field "team_id" schema + <*> (.adminlessDeletionJobConversationId) .= field "conversation_id" schema + <*> (.adminlessDeletionJobOrigUserId) .= field "orig_user_id" schema + +-- | Payload for adminless reminders. +-- Keep the JSON encoding backwards compatible. Arbiter persists these payloads +-- and workers decode them later, so changing field names or shapes without a +-- migration will break already scheduled jobs. +data AdminlessReminderJob = AdminlessReminderJob + { adminlessReminderJobTeamId :: TeamId, + adminlessReminderJobConversationId :: ConvId, + adminlessReminderJobOrigUserId :: UserId, + adminlessReminderJobDeletionScheduledFor :: UTCTimeMillis + } + deriving stock (Eq, Generic, Show) + deriving (ToJSON, FromJSON, S.ToSchema) via (Schema AdminlessReminderJob) + +instance Arbitrary AdminlessReminderJob where + arbitrary = AdminlessReminderJob <$> arbitrary <*> arbitrary <*> arbitrary <*> arbitrary + +instance ToSchema AdminlessReminderJob where + schema = + object $ + AdminlessReminderJob + <$> (.adminlessReminderJobTeamId) .= field "team_id" schema + <*> (.adminlessReminderJobConversationId) .= field "conversation_id" schema + <*> (.adminlessReminderJobOrigUserId) .= field "orig_user_id" schema + <*> (.adminlessReminderJobDeletionScheduledFor) .= field "deletion_scheduled_for" schema + +-- | Registry for the scheduled jobs we expose via Arbiter. +type ScheduledJobsRegistry = + '[ '("meetings_cleanup_jobs", MeetingsCleanupJob), + '("adminless_deletion_jobs", AdminlessDeletionJob), + '("adminless_reminder_jobs", AdminlessReminderJob) + ] diff --git a/libs/wire-api/src/Wire/API/Team/FeatureFlags.hs b/libs/wire-api/src/Wire/API/Team/FeatureFlags.hs index 6a77beeb88e..ed7e457a9a7 100644 --- a/libs/wire-api/src/Wire/API/Team/FeatureFlags.hs +++ b/libs/wire-api/src/Wire/API/Team/FeatureFlags.hs @@ -166,7 +166,11 @@ instance Default (FeatureDefaults SearchVisibilityAvailableConfig) where def = FeatureTeamSearchVisibilityAvailableByDefault instance ParseFeatureDefaults (FeatureDefaults SearchVisibilityAvailableConfig) where - parseFeatureDefaults obj = obj .: "teamSearchVisibility" + parseFeatureDefaults obj = do + -- Accept both the current feature key and the legacy team-scoped key. + mCurrent <- obj .:? "searchVisibility" + mLegacy <- obj .:? "teamSearchVisibility" + pure $ fromMaybe def (mCurrent <|> mLegacy) instance FromJSON (FeatureDefaults SearchVisibilityAvailableConfig) where parseJSON (String "enabled-by-default") = pure FeatureTeamSearchVisibilityAvailableByDefault diff --git a/libs/wire-api/test/unit/Test/Wire/API/Conversation.hs b/libs/wire-api/test/unit/Test/Wire/API/Conversation.hs index aa9527a3e69..026a393bbd3 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Conversation.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Conversation.hs @@ -75,6 +75,7 @@ testIsCellsConversationEvent = ConvReset -> isCellsConversationEvent e === False ConvMessageTimerUpdate -> isCellsConversationEvent e === False ConvHistoryUpdate -> isCellsConversationEvent e === False + ConvAdminlessReminder -> isCellsConversationEvent e === False ConvReceiptModeUpdate -> isCellsConversationEvent e === False ConvRename -> isCellsConversationEvent e === True MemberJoin -> isCellsConversationEvent e === True diff --git a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs index ab1c1aea3c6..06716dbe3b2 100644 --- a/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs +++ b/libs/wire-api/test/unit/Test/Wire/API/Roundtrip/Aeson.hs @@ -42,6 +42,7 @@ import Wire.API.Event.Conversation qualified as Event.Conversation import Wire.API.Event.Team qualified as Event.Team import Wire.API.Event.WebSocketProtocol qualified as EventWebSocketProtocol import Wire.API.FederationStatus qualified as FederationStatus +import Wire.API.Jobs qualified as Jobs import Wire.API.Locale qualified as Locale import Wire.API.Message qualified as Message import Wire.API.OAuth qualified as OAuth @@ -153,6 +154,9 @@ tests = testRoundTrip @Conversation.Role.ConversationRolesList, testRoundTrip @Conversation.Typing.TypingStatus, testRoundTrip @CustomBackend.CustomBackend, + testRoundTrip @Jobs.MeetingsCleanupJob, + testRoundTripWithSwagger @Jobs.AdminlessDeletionJob, + testRoundTripWithSwagger @Jobs.AdminlessReminderJob, testRoundTrip @EJPD.EJPDContact, testRoundTrip @Event.Conversation.Event, testRoundTrip @Event.Conversation.EventType, diff --git a/libs/wire-api/wire-api.cabal b/libs/wire-api/wire-api.cabal index 1ef9703bef6..26a5a410330 100644 --- a/libs/wire-api/wire-api.cabal +++ b/libs/wire-api/wire-api.cabal @@ -112,6 +112,7 @@ library Wire.API.History Wire.API.Internal.BulkPush Wire.API.Internal.Notification + Wire.API.Jobs Wire.API.Locale Wire.API.Meeting Wire.API.Message diff --git a/libs/wire-subsystems/default.nix b/libs/wire-subsystems/default.nix index cbda44f339a..3a6ae814248 100644 --- a/libs/wire-subsystems/default.nix +++ b/libs/wire-subsystems/default.nix @@ -11,6 +11,12 @@ , amazonka-ses , amazonka-sqs , amqp +, arbiter-core +, arbiter-hasql +, arbiter-migrations +, arbiter-worker +, asn1-encoding +, asn1-types , async , attoparsec , base @@ -30,6 +36,7 @@ , contravariant , cookie , cql +, cron , crypton , crypton-asn1-encoding , crypton-asn1-types @@ -152,6 +159,12 @@ mkDerivation { amazonka-ses amazonka-sqs amqp + arbiter-core + arbiter-hasql + arbiter-migrations + arbiter-worker + asn1-encoding + asn1-types async attoparsec base @@ -171,6 +184,7 @@ mkDerivation { contravariant cookie cql + cron crypton crypton-asn1-encoding crypton-asn1-types @@ -281,6 +295,12 @@ mkDerivation { amazonka-ses amazonka-sqs amqp + arbiter-core + arbiter-hasql + arbiter-migrations + arbiter-worker + asn1-encoding + asn1-types async attoparsec base @@ -299,6 +319,7 @@ mkDerivation { contravariant cookie cql + cron crypton crypton-asn1-encoding crypton-asn1-types diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem.hs index d1991691258..a6bde14a526 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem.hs @@ -32,6 +32,7 @@ import Data.Code qualified as Code import Data.CommaSeparatedList (CommaSeparatedList) import Data.Domain import Data.Id +import Data.Json.Util (UTCTimeMillis) import Data.Misc (IpAddr) import Data.Qualified import Data.Range @@ -341,6 +342,15 @@ data ConversationSubsystem m a where InternalDeleteLocalConversation :: Local ConvId -> ConversationSubsystem m () + InternalDeleteLocalAdminlessGroup :: + Local UserId -> + Local ConvId -> + ConversationSubsystem m () + InternalNotifyAdminlessReminder :: + Local UserId -> + Local ConvId -> + UTCTimeMillis -> + ConversationSubsystem m () GetMLSPublicKeys :: Maybe MLSPublicKeyFormat -> ConversationSubsystem m (MLSKeysByPurpose (MLSKeys SomeKey)) diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action.hs index a6e6921db12..d4ebac964ce 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Action.hs @@ -51,6 +51,7 @@ module Wire.ConversationSubsystem.Action addLocalUsersToRemoteConv, ConversationUpdate, ensureAllowed, + removeConversation, ) where @@ -115,6 +116,7 @@ import Wire.BrigAPIAccess qualified as E import Wire.CodeStore import Wire.CodeStore qualified as E import Wire.CodeStore.Code (CodeReferent (..)) +import Wire.ConversationStore (ConversationStore) import Wire.ConversationStore qualified as E import Wire.ConversationSubsystem.Action.Kick import Wire.ConversationSubsystem.Action.Leave @@ -375,28 +377,7 @@ instance IsConversationAction 'ConversationDeleteTag where ] performAction lconv _qusr _conId () = do - let lcnv = fmap (.id_) lconv - storedConv = tUnqualified lconv - let deleteGroup groupId = do - E.removeAllMLSClients groupId - E.removeAllHistoryClients groupId - E.deleteAllProposals groupId - - let cid = storedConv.id_ - for_ (storedConv & mlsMetadata <&> cnvmlsGroupId . fst) $ \gidParent -> do - sconvs <- E.listSubConversations cid - for_ (Map.assocs sconvs) $ \(subid, mlsData) -> do - let gidSub = cnvmlsGroupId mlsData - E.deleteSubConversation cid subid - deleteGroup gidSub - deleteGroup gidParent - - key <- E.makeKey (CodeReferentConv (tUnqualified lcnv)) - E.deleteCode key - case convTeam storedConv of - Nothing -> E.deleteConversation (tUnqualified lcnv) - Just tid -> E.deleteTeamConversation tid (tUnqualified lcnv) - + removeConversation lconv pure $ mkPerformActionResult () ensureAllowed _ _action _conv (ActorContext Nothing (Just _tm)) = @@ -409,6 +390,36 @@ instance IsConversationAction 'ConversationDeleteTag where allowChannelManagePermission = True +removeConversation :: + ( Member ConversationStore r, + Member ProposalStore r, + Member CodeStore r + ) => + Local StoredConversation -> + Sem r () +removeConversation lconv = do + let lcnv = fmap (.id_) lconv + storedConv = tUnqualified lconv + let deleteGroup groupId = do + E.removeAllMLSClients groupId + E.removeAllHistoryClients groupId + E.deleteAllProposals groupId + + let cid = storedConv.id_ + for_ (storedConv & mlsMetadata <&> cnvmlsGroupId . fst) $ \gidParent -> do + sconvs <- E.listSubConversations cid + for_ (Map.assocs sconvs) $ \(subid, mlsData) -> do + let gidSub = cnvmlsGroupId mlsData + E.deleteSubConversation cid subid + deleteGroup gidSub + deleteGroup gidParent + + key <- E.makeKey (CodeReferentConv (tUnqualified lcnv)) + E.deleteCode key + case convTeam storedConv of + Nothing -> E.deleteConversation (tUnqualified lcnv) + Just tid -> E.deleteTeamConversation tid (tUnqualified lcnv) + instance IsConversationAction 'ConversationRenameTag where type HasConversationActionEffects 'ConversationRenameTag r = diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Interpreter.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Interpreter.hs index 972ec7573b8..884e707d313 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Interpreter.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Interpreter.hs @@ -71,6 +71,7 @@ import Wire.FederationAPIAccess (FederationAPIAccess) import Wire.FederationSubsystem (FederationSubsystem) import Wire.FireAndForget (FireAndForget) import Wire.HashPassword (HashPassword) +import Wire.JobSubsystem (JobSubsystem) import Wire.LegalHoldStore (LegalHoldStore) import Wire.NotificationSubsystem as NS import Wire.Options.Galley (GuestLinkTTLSeconds) @@ -115,6 +116,7 @@ interpretConversationSubsystem :: Member TeamStore r, Member ConvStore.MLSCommitLockStore r, Member FederationSubsystem r, + Member JobSubsystem r, Member Resource r, Member (Input (Maybe (MLSKeysByPurpose MLSPrivateKeys))) r, Member UserClientIndexStore r, @@ -211,6 +213,10 @@ interpretConversationSubsystem = interpret $ \case mapErrors $ Update.deleteLocalConversation lusr con lcnv InternalDeleteLocalConversation lcnv -> mapErrors $ Action.updateLocalConversationDeleteUnchecked lcnv + InternalDeleteLocalAdminlessGroup lusr lcnv -> + mapErrors $ Update.adminlessAutopromoteOrDelete lusr lcnv + InternalNotifyAdminlessReminder lusr lcnv deletionScheduledFor -> + mapErrors $ Update.adminlessAutopromoteOrSendReminder lusr lcnv deletionScheduledFor GetMLSPublicKeys fmt -> mapErrors $ MLS.getMLSPublicKeys fmt ResetMLSConversation lusr reset -> diff --git a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Update.hs b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Update.hs index a96558746bd..2c7e41ce082 100644 --- a/libs/wire-subsystems/src/Wire/ConversationSubsystem/Update.hs +++ b/libs/wire-subsystems/src/Wire/ConversationSubsystem/Update.hs @@ -42,6 +42,8 @@ module Wire.ConversationSubsystem.Update updateConversationProtocolWithLocalUser, updateLocalStateOfRemoteConv, updateCellsState, + adminlessAutopromoteOrDelete, + adminlessAutopromoteOrSendReminder, -- * Managing Members addQualifiedMembersUnqualified, @@ -84,6 +86,7 @@ import Data.Misc import Data.Qualified import Data.Set qualified as Set import Data.Singletons +import Data.Time.Clock (NominalDiffTime, addUTCTime) import Data.Vector qualified as V import Galley.Types.Error import Imports hiding (forkIO) @@ -140,6 +143,7 @@ import Wire.FederationAPIAccess qualified as E import Wire.FederationSubsystem import Wire.FireAndForget import Wire.HashPassword as HashPassword +import Wire.JobSubsystem (JobSubsystem, scheduleAdminlessDeletionJob, scheduleAdminlessReminderJob) import Wire.LegalHoldStore (LegalHoldStore) import Wire.NotificationSubsystem import Wire.Options.Galley @@ -972,6 +976,7 @@ replaceMembers :: Member FederationSubsystem r, Member FeaturesConfigSubsystem r, Member TeamSubsystem r, + Member JobSubsystem r, Member (Input ConversationSubsystemConfig) r ) => Local UserId -> @@ -1153,6 +1158,7 @@ removeMemberQualified :: Member TinyLog r, Member FeaturesConfigSubsystem r, Member TeamSubsystem r, + Member JobSubsystem r, Member (Input ConversationSubsystemConfig) r ) => RemoveMemberResponseMode -> @@ -1185,7 +1191,8 @@ guardPreventAdminlessGroups :: Member E.ExternalAccess r, Member BackendNotificationQueueAccess r, Member TeamSubsystem r, - Member FeaturesConfigSubsystem r + Member FeaturesConfigSubsystem r, + Member JobSubsystem r ) => RemoveMemberResponseMode -> Local ConvId -> @@ -1197,7 +1204,7 @@ guardPreventAdminlessGroups responseMode lcnv lusr victim = do for_ conv.metadata.cnvmTeam $ \tid -> do (feature :: LockableFeature PreventAdminlessGroupsConfig) <- getFeatureForTeam tid when (feature.status == FeatureStatusEnabled && isLeavingLastConversationAdmin (qUnqualified victim) conv) $ do - eligibleMembers <- eligibleAdminFallbackMembers lcnv (qUnqualified victim) conv + eligibleMembers <- eligibleAdminFallbackMembers lcnv (Just (qUnqualified victim)) conv case (responseMode, eligibleMembers) of (RemoveMemberLegacyResponse, x : xs) -> do seed <- randomWord64 @@ -1208,18 +1215,159 @@ guardPreventAdminlessGroups responseMode lcnv lusr victim = do (RemoveMemberEligibleMembersResponse, _ : _) -> throw $ AdminlessConversation (fmap fst eligibleMembers) (RemoveMemberLegacyResponse, []) -> - -- FUTUREWORK: mark for deletion - pure () - (RemoveMemberEligibleMembersResponse, []) -> do - -- FUTUREWORK: mark for deletion - pure () + scheduleDeletion tid feature + (RemoveMemberEligibleMembersResponse, []) -> + scheduleDeletion tid feature + where + scheduleDeletion tid feature = do + now <- Now.get + let deletionTimeout = timeoutToNominalDiffTime feature.config.deletionTimeout + scheduledFor = addUTCTime deletionTimeout now + deletionScheduledFor = toUTCTimeMillis scheduledFor + void $ scheduleAdminlessDeletionJob lusr tid (qUnqualified (tUntagged lcnv)) scheduledFor + for_ feature.config.reminderTimeouts $ + scheduleReminder now tid deletionScheduledFor deletionTimeout + + scheduleReminder now tid deletionScheduledFor deletionTimeout reminderTimeoutCfg = do + let reminderTimeout = timeoutToNominalDiffTime reminderTimeoutCfg + when (reminderTimeout < deletionTimeout) $ do + let reminderAt = addUTCTime (deletionTimeout - reminderTimeout) now + void $ + scheduleAdminlessReminderJob + lusr + tid + (qUnqualified (tUntagged lcnv)) + deletionScheduledFor + reminderTimeout + reminderAt + + timeoutToNominalDiffTime :: PreventAdminlessTimeout -> NominalDiffTime + timeoutToNominalDiffTime = + realToFrac . duration . durationLiteralValue . preventAdminlessTimeoutLiteral + +onAdminless :: + ( Member ConversationStore r, + Member (ErrorS 'ConvNotFound) r, + Member BrigAPIAccess r, + Member FeaturesConfigSubsystem r + ) => + Local ConvId -> + (StoredConversation -> LockableFeature PreventAdminlessGroupsConfig -> [(Qualified UserId, User.Name)] -> Sem r ()) -> + Sem r () +onAdminless lcnv action = do + conv <- getConversationWithError lcnv + for_ conv.metadata.cnvmTeam $ \tid -> do + (feature :: LockableFeature PreventAdminlessGroupsConfig) <- getFeatureForTeam tid + let adminExists = any (\member -> member.convRoleName == roleNameWireAdmin) conv.localMembers || any (\member -> member.convRoleName == roleNameWireAdmin) conv.remoteMembers + when (feature.status == FeatureStatusEnabled && not adminExists) $ do + eligibleMembers <- eligibleAdminFallbackMembers lcnv Nothing conv + action conv feature eligibleMembers + +adminlessTryAutopromote :: + ( Member ConversationStore r, + Member (ErrorS 'ConvNotFound) r, + Member (Error FederationError) r, + Member BrigAPIAccess r, + Member Random r, + Member NotificationSubsystem r, + Member Now r, + Member E.ExternalAccess r, + Member BackendNotificationQueueAccess r, + Member FeaturesConfigSubsystem r + ) => + Local UserId -> + Local ConvId -> + (StoredConversation -> Sem r ()) -> + Sem r () +adminlessTryAutopromote lusr lcnv altAction = do + onAdminless lcnv $ \conv feature eligibleMembers -> do + case eligibleMembers of + x : xs -> do + seed <- randomWord64 + let autopromotionCandidates = selectAutopromotionCandidate seed feature.config.promotionStrategy (x :| xs) + update = (OtherMemberUpdate (Just roleNameWireAdmin)) + for_ autopromotionCandidates $ \candidate -> do + E.setOtherMember lcnv candidate update + void $ + sendConversationActionNotifications + (sing @'ConversationMemberUpdateTag) + (tUntagged lusr) + False + Nothing + (qualifyAs lcnv conv) + (convBotsAndMembers conv) + (ConversationMemberUpdate (tUntagged lusr) update) + def + [] -> altAction conv + +adminlessAutopromoteOrDelete :: + ( Member ConversationStore r, + Member (ErrorS 'ConvNotFound) r, + Member (Error FederationError) r, + Member BrigAPIAccess r, + Member Random r, + Member NotificationSubsystem r, + Member Now r, + Member E.ExternalAccess r, + Member BackendNotificationQueueAccess r, + Member FeaturesConfigSubsystem r, + Member ProposalStore r, + Member CodeStore r + ) => + Local UserId -> + Local ConvId -> + Sem r () +adminlessAutopromoteOrDelete lusr lcnv = adminlessTryAutopromote lusr lcnv orAlternativelyDeleteConv + where + orAlternativelyDeleteConv conv = do + removeConversation (qualifyAs lcnv conv) + void $ + sendConversationActionNotifications + (sing @'ConversationDeleteTag) + (tUntagged lusr) + False + Nothing + (qualifyAs lcnv conv) + (convBotsAndMembers conv) + () + def + +adminlessAutopromoteOrSendReminder :: + ( Member ConversationStore r, + Member (ErrorS 'ConvNotFound) r, + Member (Error FederationError) r, + Member BrigAPIAccess r, + Member Random r, + Member NotificationSubsystem r, + Member Now r, + Member E.ExternalAccess r, + Member BackendNotificationQueueAccess r, + Member FeaturesConfigSubsystem r + ) => + Local UserId -> + Local ConvId -> + UTCTimeMillis -> + Sem r () +adminlessAutopromoteOrSendReminder lusr lcnv deletionScheduledFor = adminlessTryAutopromote lusr lcnv orAlternativelySendReminder + where + orAlternativelySendReminder conv = do + now <- Now.get + let event = + Event + (tUntagged lcnv) + Nothing + (EventFromUser (tUntagged lusr)) + now + (conv.metadata.cnvmTeam) + (EdAdminlessReminder (AdminlessReminder deletionScheduledFor)) + pushConversationEvent Nothing conv event (qualifyAs lcnv (map (.id_) conv.localMembers)) [] + +-- Use eight random bytes and fold them into a big-endian Word64. This keeps +-- the helper small, deterministic under tests, and free of extra Random API. +randomWord64 :: (Member Random r) => Sem r Word64 +randomWord64 = BS.foldl' step 0 <$> Random.bytes 8 where - -- Use eight random bytes and fold them into a big-endian Word64. This keeps - -- the helper small, deterministic under tests, and free of extra Random API. - randomWord64 :: (Member Random r) => Sem r Word64 - randomWord64 = BS.foldl' step 0 <$> Random.bytes 8 - where - step acc byte = shiftL acc 8 .|. fromIntegral byte + step acc byte = shiftL acc 8 .|. fromIntegral byte isLeavingLastConversationAdmin :: UserId -> StoredConversation -> Bool isLeavingLastConversationAdmin leavingUser conv = @@ -1234,16 +1382,16 @@ isLeavingLastConversationAdmin leavingUser conv = eligibleAdminFallbackMembers :: (Member BrigAPIAccess r) => Local ConvId -> - UserId -> + Maybe UserId -> StoredConversation -> Sem r [(Qualified UserId, User.Name)] -eligibleAdminFallbackMembers lcnv leavingUser conv = do - users <- Brig.getUsers (map (.id_) (filter ((/= leavingUser) . (.id_)) conv.localMembers)) +eligibleAdminFallbackMembers lcnv mLeavingUser conv = do + users <- Brig.getUsers (map (.id_) (filter ((/= mLeavingUser) . Just . (.id_)) conv.localMembers)) let usersById = Map.fromList [(User.userId u, u) | u <- users] pure [ (tUntagged (qualifyAs lcnv member.id_), u.userDisplayName) | member <- conv.localMembers, - member.id_ /= leavingUser, + Just member.id_ /= mLeavingUser, Just u <- [Map.lookup member.id_ usersById], isEligibleUser u ] @@ -1268,7 +1416,8 @@ deleteUserFromTeamConversationsImpl :: Member E.ExternalAccess r, Member Now r, Member Random r, - Member TeamSubsystem r + Member TeamSubsystem r, + Member JobSubsystem r ) => Local UserId -> Maybe ConnId -> @@ -1379,7 +1528,8 @@ removeMemberFromLocalConv :: Member (ErrorS ConvMemberNotFound) r, Member (Error AdminlessConversation) r, Member FeaturesConfigSubsystem r, - Member BrigAPIAccess r + Member BrigAPIAccess r, + Member JobSubsystem r ) => RemoveMemberResponseMode -> Local ConvId -> diff --git a/libs/wire-subsystems/src/Wire/JobSubsystem.hs b/libs/wire-subsystems/src/Wire/JobSubsystem.hs new file mode 100644 index 00000000000..1967421c346 --- /dev/null +++ b/libs/wire-subsystems/src/Wire/JobSubsystem.hs @@ -0,0 +1,65 @@ +{-# LANGUAGE TemplateHaskell #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License +-- for more details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.JobSubsystem + ( CleanupAction, + JobSubsystemConfig (..), + JobSubsystem (..), + JobWorkerHandlers (..), + JobWorkersConfig (..), + scheduleAdminlessDeletionJob, + scheduleAdminlessReminderJob, + startJobWorkers, + ) +where + +import Arbiter.Core.Job.Types (JobRead) +import Data.Id +import Data.Json.Util (UTCTimeMillis) +import Data.Qualified +import Data.Time.Clock (NominalDiffTime, UTCTime) +import Imports +import Polysemy +import Wire.API.Jobs +import Wire.JobSubsystem.Workers + +type CleanupAction = IO () + +data JobSubsystemConfig = JobSubsystemConfig + { jobSubsystemSchemaName :: Text + } + +data JobWorkersConfig = JobWorkersConfig + { recurringJobRunnerConfig :: RecurringJobRunnerConfig ScheduledJobsRegistry, + adminlessDeletionJobRunnerConfig :: OneOffJobRunnerConfig ScheduledJobsRegistry AdminlessDeletionJob, + adminlessReminderJobRunnerConfig :: OneOffJobRunnerConfig ScheduledJobsRegistry AdminlessReminderJob + } + +data JobWorkerHandlers = JobWorkerHandlers + { recurringJobRunnerRunJob :: MeetingsCleanupJob -> IO (), + adminlessDeletionJobRunnerRunJob :: JobRead AdminlessDeletionJob -> IO (), + adminlessReminderJobRunnerRunJob :: JobRead AdminlessReminderJob -> IO () + } + +data JobSubsystem m a where + ScheduleAdminlessDeletionJob :: Local UserId -> TeamId -> ConvId -> UTCTime -> JobSubsystem m () + ScheduleAdminlessReminderJob :: Local UserId -> TeamId -> ConvId -> UTCTimeMillis -> NominalDiffTime -> UTCTime -> JobSubsystem m () + StartJobWorkers :: JobWorkersConfig -> JobWorkerHandlers -> JobSubsystem m CleanupAction + +makeSem ''JobSubsystem diff --git a/libs/wire-subsystems/src/Wire/JobSubsystem/ArbiterAdapter.hs b/libs/wire-subsystems/src/Wire/JobSubsystem/ArbiterAdapter.hs new file mode 100644 index 00000000000..d2b30870f6e --- /dev/null +++ b/libs/wire-subsystems/src/Wire/JobSubsystem/ArbiterAdapter.hs @@ -0,0 +1,178 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE TypeFamilies #-} + +-- | Adapter that lets Arbiter run against wire-server's shared Hasql pool. +-- +-- The pool we want to reuse is 'HasqlPoolExt.Pool'. Internally that is backed +-- by @Data.Pool (Either ConnectionError Connection)@, but the underlying +-- resource pool is intentionally opaque in @hasql-resource-pool@. The only +-- missing piece is therefore a small exported helper from that package that +-- borrows one live 'Connection' for the duration of a callback. +module Wire.JobSubsystem.ArbiterAdapter where + +import Arbiter.Core.Codec (Params, RowCodec) +import Arbiter.Core.Exceptions (throwInternal) +import Arbiter.Core.HasArbiterSchema (HasArbiterSchema (..)) +import Arbiter.Core.MonadArbiter (MonadArbiter (..)) +import Arbiter.Core.QueueRegistry (JobPayloadRegistry) +import Arbiter.Hasql.Decode qualified as Decode +import Arbiter.Hasql.Encode qualified as Encode +import Control.Exception (mask, onException, try) +import Control.Monad.Catch (MonadCatch, MonadMask, MonadThrow) +import Control.Monad.Reader +import Data.Misc (durationToCeilingSeconds) +import Data.Text qualified as T +import Hasql.Connection qualified as HasqlConn +import Hasql.Decoders qualified as Decoders +import Hasql.Encoders qualified as Encoders +import Hasql.Pool qualified as HasqlPool +import Hasql.Pool.Extended qualified as HasqlPoolExt +import Hasql.Session qualified as Session +import Hasql.Statement qualified as Statement +import Imports + +data WireArbiterEnv = WireArbiterEnv + { schemaName :: Text, + connectionPool :: HasqlPoolExt.Pool, + activeConn :: Maybe HasqlConn.Connection, + transactionDepth :: Int, + preparedStatements :: Bool + } + +newtype WireArbiter (registry :: JobPayloadRegistry) a = WireArbiter + { unWireArbiter :: ReaderT WireArbiterEnv IO a + } + deriving newtype + ( Functor, + Applicative, + Monad, + MonadCatch, + MonadIO, + MonadMask, + MonadReader WireArbiterEnv, + MonadThrow, + MonadUnliftIO + ) + +runWireArbiter :: WireArbiterEnv -> WireArbiter registry a -> IO a +runWireArbiter env (WireArbiter action) = runReaderT action env + +instance HasArbiterSchema (WireArbiter registry) registry where + getSchema = asks schemaName + +instance MonadArbiter (WireArbiter registry) where + type Handler (WireArbiter registry) jobs result = HasqlConn.Connection -> jobs -> WireArbiter registry result + + executeQuery sql params codec = do + env <- ask + withConn env $ \conn -> + runQueryStatement env.preparedStatements conn sql params codec + + executeStatement sql params = do + env <- ask + withConn env $ \conn -> + runExecStatement conn sql params + + withDbTransaction (WireArbiter action) = WireArbiter $ do + env <- ask + case activeConn env of + Nothing -> + unWireArbiter $ + withPoolConnection env.connectionPool $ \conn -> + beginCommitOrRollback conn $ + runReaderT action env {activeConn = Just conn, transactionDepth = 1} + Just conn + | transactionDepth env <= 0 -> + liftIO $ + beginCommitOrRollback conn $ + runReaderT action env {transactionDepth = 1} + | otherwise -> + liftIO $ + beginSavepointTransaction (transactionDepth env) conn $ + runReaderT action env {transactionDepth = transactionDepth env + 1} + + runHandlerWithConnection handler jobs = do + env <- ask + case activeConn env of + Just conn -> handler conn jobs + Nothing -> throwInternal "runHandlerWithConnection: no active connection" + +withConn :: WireArbiterEnv -> (HasqlConn.Connection -> IO a) -> WireArbiter registry a +withConn env f = + case activeConn env of + Just conn -> liftIO $ f conn + Nothing -> withPoolConnection env.connectionPool f + +-- | Borrow a live connection from wire-server's shared pool. +withPoolConnection :: HasqlPoolExt.Pool -> (HasqlConn.Connection -> IO a) -> WireArbiter registry a +withPoolConnection pool f = do + result <- + liftIO $ + HasqlPool.withConnectionWithPoolAcquisitionTimeout + (durationToCeilingSeconds pool.poolAcquisitionTimeout) + pool.rawPool + (fmap Right . f) + case result of + Right x -> pure x + Left HasqlPool.AcquisitionTimeoutUsageError -> do + liftIO $ HasqlPoolExt.recordHasqlPoolAcquisitionTimeout pool.metrics + throwInternal "hasql pool acquisition timeout" + Left (HasqlPool.ConnectionError err) -> do + liftIO $ HasqlPoolExt.recordHasqlPoolConnectionFailure pool.metrics + throwInternal $ "hasql connection error: " <> T.pack (show err) + Left (HasqlPool.SessionError err) -> do + liftIO $ HasqlPoolExt.recordHasqlPoolSessionFailure pool.metrics + throwInternal $ "hasql session error: " <> T.pack (show err) + +runQueryStatement :: Bool -> HasqlConn.Connection -> Text -> Params -> RowCodec a -> IO [a] +runQueryStatement prepare conn sql params codec = do + let mk = if prepare then Statement.preparable else Statement.unpreparable + stmt = + mk + (Encode.convertPlaceholders sql) + (Encode.buildEncoder params) + (Decode.hasqlRowDecoder codec) + result <- HasqlConn.use conn (Session.statement () stmt) + case result of + Right rows -> pure rows + Left err -> throwInternal $ "hasql query error: " <> T.pack (show err) + +runExecStatement :: HasqlConn.Connection -> Text -> Params -> IO Int64 +runExecStatement conn sql params = do + let stmt = Encode.buildStatementRowCount sql params + result <- HasqlConn.use conn (Session.statement () stmt) + case result of + Right n -> pure n + Left err -> throwInternal $ "hasql statement error: " <> T.pack (show err) + +runRawSql :: HasqlConn.Connection -> Text -> IO () +runRawSql conn sql = do + let stmt = Statement.unpreparable sql Encoders.noParams Decoders.noResult + result <- HasqlConn.use conn (Session.statement () stmt) + case result of + Right () -> pure () + Left err -> throwInternal $ "hasql sql error: " <> T.pack (show err) + +beginCommitOrRollback :: HasqlConn.Connection -> IO a -> IO a +beginCommitOrRollback conn action = mask $ \restore -> do + runRawSql conn "BEGIN" + result <- restore action `onException` rollbackSafely + runRawSql conn "COMMIT" + pure result + where + rollbackSafely = do + _ <- try (runRawSql conn "ROLLBACK") :: IO (Either SomeException ()) + pure () + +beginSavepointTransaction :: Int -> HasqlConn.Connection -> IO a -> IO a +beginSavepointTransaction depth conn action = mask $ \restore -> do + let spName = "arbiter_sp_" <> T.pack (show depth) + runRawSql conn ("SAVEPOINT " <> spName) + result <- + restore action + `onException` do + _ <- try (runRawSql conn ("ROLLBACK TO SAVEPOINT " <> spName)) :: IO (Either SomeException ()) + pure () + runRawSql conn ("RELEASE SAVEPOINT " <> spName) + pure result diff --git a/libs/wire-subsystems/src/Wire/JobSubsystem/Interpreter.hs b/libs/wire-subsystems/src/Wire/JobSubsystem/Interpreter.hs new file mode 100644 index 00000000000..5bcf62c148c --- /dev/null +++ b/libs/wire-subsystems/src/Wire/JobSubsystem/Interpreter.hs @@ -0,0 +1,135 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeFamilies #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License +-- for more details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.JobSubsystem.Interpreter + ( interpretJobSubsystem, + runJobWorkers, + ) +where + +import Arbiter.Core qualified as ArbiterCore +import Data.Id +import Data.Json.Util (UTCTimeMillis) +import Data.Qualified +import Data.Text qualified as Text +import Data.Time +import Hasql.Pool.Extended qualified as HasqlPoolExt +import Imports +import Polysemy +import Polysemy.Input (input) +import Wire.API.Jobs +import Wire.JobSubsystem (CleanupAction, JobSubsystem (..), JobSubsystemConfig (..), JobWorkerHandlers (..), JobWorkersConfig (..)) +import Wire.JobSubsystem.ArbiterAdapter (WireArbiter, WireArbiterEnv (..), runWireArbiter) +import Wire.JobSubsystem.Workers + ( RecurringJobRunnerConfig (..), + runOneOffJobRunner, + runRecurringJobRunner, + runScheduledJobsMigrations, + ) +import Wire.Postgres (PGConstraints) + +runJobWorkers :: HasqlPoolExt.Pool -> JobWorkersConfig -> JobWorkerHandlers -> IO CleanupAction +runJobWorkers pool JobWorkersConfig {..} JobWorkerHandlers {..} = do + runScheduledJobsMigrations + (recurringJobRunnerArbiterConnStr recurringJobRunnerConfig) + (recurringJobRunnerSchemaName recurringJobRunnerConfig) + cleanupRecurring <- runRecurringJobRunner @ScheduledJobsRegistry pool recurringJobRunnerConfig recurringJobRunnerRunJob + cleanupDeletion <- runOneOffJobRunner @ScheduledJobsRegistry pool adminlessDeletionJobRunnerConfig adminlessDeletionJobRunnerRunJob + cleanupReminder <- runOneOffJobRunner @ScheduledJobsRegistry pool adminlessReminderJobRunnerConfig adminlessReminderJobRunnerRunJob + pure $ cleanupRecurring >> cleanupDeletion >> cleanupReminder + +interpretJobSubsystem :: + (PGConstraints r) => + JobSubsystemConfig -> + InterpreterFor JobSubsystem r +interpretJobSubsystem conf = + interpret + \case + ScheduleAdminlessDeletionJob lusr tid cid scheduledFor -> scheduleAdminlessDeletionJob conf lusr tid cid scheduledFor + ScheduleAdminlessReminderJob lusr tid cid deletionScheduledFor reminderTimeout scheduledFor -> scheduleAdminlessReminderJob conf lusr tid cid deletionScheduledFor reminderTimeout scheduledFor + StartJobWorkers cfg handlers -> do + pool <- input + embed $ runJobWorkers pool cfg handlers + +scheduleAdminlessDeletionJob :: + forall r. + (PGConstraints r) => + JobSubsystemConfig -> + Local UserId -> + TeamId -> + ConvId -> + UTCTime -> + Sem r () +scheduleAdminlessDeletionJob JobSubsystemConfig {..} lusr teamId convId scheduledFor = do + pool <- input + let arbiterEnv = + WireArbiterEnv + { schemaName = jobSubsystemSchemaName, + connectionPool = pool, + activeConn = Nothing, + transactionDepth = 0, + preparedStatements = False + } + let arbiterJob = + (ArbiterCore.defaultGroupedJob adminlessDeletionQueueName (AdminlessDeletionJob teamId convId (tUnqualified lusr))) + { ArbiterCore.notVisibleUntil = Just scheduledFor, + ArbiterCore.dedupKey = Just . ArbiterCore.IgnoreDuplicate $ adminlessJobDedupKey "deletion" teamId convId, + ArbiterCore.maxAttempts = Just 3 + } + embed $ void $ runWireArbiter arbiterEnv $ ArbiterCore.insertJob @(WireArbiter ScheduledJobsRegistry) arbiterJob + +scheduleAdminlessReminderJob :: + forall r. + (PGConstraints r) => + JobSubsystemConfig -> + Local UserId -> + TeamId -> + ConvId -> + UTCTimeMillis -> + NominalDiffTime -> + UTCTime -> + Sem r () +scheduleAdminlessReminderJob JobSubsystemConfig {..} lusr teamId convId deletionScheduledFor reminderTimeout scheduledFor = do + pool <- input @HasqlPoolExt.Pool + let arbiterEnv = + WireArbiterEnv + { schemaName = jobSubsystemSchemaName, + connectionPool = pool, + activeConn = Nothing, + transactionDepth = 0, + preparedStatements = False + } + let arbiterJob = + (ArbiterCore.defaultGroupedJob adminlessReminderQueueName (AdminlessReminderJob teamId convId (tUnqualified lusr) deletionScheduledFor)) + { ArbiterCore.notVisibleUntil = Just scheduledFor, + ArbiterCore.dedupKey = Just . ArbiterCore.IgnoreDuplicate $ adminlessReminderJobDedupKey teamId convId reminderTimeout, + ArbiterCore.maxAttempts = Just 3 + } + embed $ void $ runWireArbiter arbiterEnv $ ArbiterCore.insertJob @(WireArbiter ScheduledJobsRegistry) arbiterJob + +adminlessJobDedupKey :: Text -> TeamId -> ConvId -> Text +adminlessJobDedupKey jobType teamId convId = + "adminless-" <> jobType <> ":" <> idToText teamId <> ":" <> idToText convId + +adminlessReminderJobDedupKey :: TeamId -> ConvId -> NominalDiffTime -> Text +adminlessReminderJobDedupKey teamId convId reminderTimeout = + adminlessJobDedupKey "reminder" teamId convId <> ":" <> Text.pack (show reminderTimeout) diff --git a/libs/wire-subsystems/src/Wire/JobSubsystem/Workers.hs b/libs/wire-subsystems/src/Wire/JobSubsystem/Workers.hs new file mode 100644 index 00000000000..9b322a628dc --- /dev/null +++ b/libs/wire-subsystems/src/Wire/JobSubsystem/Workers.hs @@ -0,0 +1,358 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE RecordWildCards #-} +{-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE TypeApplications #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License +-- for more details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.JobSubsystem.Workers + ( RecurringJobRunnerConfig (..), + OneOffJobRunnerConfig (..), + runScheduledJobsMigrations, + runRecurringJobRunner, + runOneOffJobRunner, + ) +where + +import Arbiter.Core qualified as ArbiterCore +import Arbiter.Core.Job.Types (JobRead) +import Arbiter.Core.QueueRegistry (RegistryTables, TableForPayload) +import Arbiter.Migrations qualified as ArbiterMigrations +import Arbiter.Worker qualified as ArbiterWorker +import Arbiter.Worker.Config qualified as ArbiterWorkerConfig +import Arbiter.Worker.Cron qualified as ArbiterWorkerCron +import Control.Exception (bracket, bracket_, throwIO) +import Data.Aeson (FromJSON, ToJSON) +import Data.Hashable qualified as Hashable +import Data.Kind (Type) +import Data.Proxy (Proxy (..)) +import Data.Secret (SecretText, revealSecretText) +import Data.Text qualified as T +import Data.Text.Encoding qualified as Text +import Data.Time.Clock (NominalDiffTime) +import GHC.TypeLits (KnownSymbol) +import Hasql.Connection qualified as HasqlConnection +import Hasql.Connection.Settings qualified as HasqlConnectionSettings +import Hasql.Pool.Extended qualified as HasqlPoolExt +import Hasql.Session qualified as HasqlSession +import Hasql.Statement qualified as HasqlStatement +import Hasql.TH +import Imports +import System.Cron (CronSchedule, serializeCronSchedule) +import System.IO.Error (userError) +import System.Logger qualified as Log +import System.Timeout (timeout) +import UnliftIO.Async qualified as Async +import Wire.API.Jobs (MeetingsCleanupJob (..), ScheduledJobsRegistry) +import Wire.JobSubsystem.ArbiterAdapter (WireArbiter, WireArbiterEnv (..), runWireArbiter) + +data RecurringJobRunnerConfig registry = RecurringJobRunnerConfig + { recurringJobRunnerLogger :: Log.Logger, + recurringJobRunnerSchedule :: CronSchedule, + -- May contain the PostgreSQL password. Keep it wrapped until the Arbiter boundary. + recurringJobRunnerArbiterConnStr :: SecretText, + recurringJobRunnerSchemaName :: Text, + recurringJobRunnerPollInterval :: NominalDiffTime, + recurringJobRunnerWorkerThreads :: Int, + recurringJobRunnerJobName :: Text, + recurringJobRunnerQueueName :: Text + } + +data OneOffJobRunnerConfig registry (payload :: Type) = OneOffJobRunnerConfig + { oneOffJobRunnerLogger :: Log.Logger, + -- May contain the PostgreSQL password. Keep it wrapped until the Arbiter boundary. + oneOffJobRunnerArbiterConnStr :: SecretText, + oneOffJobRunnerSchemaName :: Text, + oneOffJobRunnerPollInterval :: NominalDiffTime, + oneOffJobRunnerWorkerThreads :: Int, + oneOffJobRunnerJobName :: Text, + oneOffJobRunnerQueueName :: Text + } + +-- | Apply all migrations for the scheduled-jobs registry before constructing +-- any worker pools or accepting scheduled jobs. +runScheduledJobsMigrations :: SecretText -> Text -> IO () +runScheduledJobsMigrations connStr schemaName = + withArbiterMigrationLock connStr schemaName $ do + result <- + ArbiterMigrations.runMigrationsForRegistry + (Proxy @ScheduledJobsRegistry) + (Text.encodeUtf8 $ revealSecretText connStr) + schemaName + ArbiterMigrations.defaultMigrationConfig + case result of + ArbiterMigrations.MigrationSuccess -> pure () + ArbiterMigrations.MigrationError err -> + throwIO . userError $ + "Arbiter migrations failed for schema " <> T.unpack schemaName <> ": " <> err + +-- | Serialize Arbiter schema migrations across all background-worker instances. +-- +-- Arbiter's migration API opens its own PostgreSQL connection, so the lock is +-- held on a separate dedicated connection for the entire migration call. This +-- is sufficient because every Wire Server migration caller enters through this +-- wrapper. The connection is kept open until the lock is released; otherwise a +-- session-level advisory lock would be released before the migrations finish. +withArbiterMigrationLock :: SecretText -> Text -> IO a -> IO a +withArbiterMigrationLock connStr schemaName action = do + bracket acquireConnection HasqlConnection.release $ \lockConnection -> do + bracket_ + (acquireArbiterMigrationLockWithTimeout lockConnection) + (runAdvisoryLockStatement lockConnection releaseArbiterMigrationLock) + action + where + lockId :: Int64 + lockId = fromIntegral . Hashable.hash $ ("wire-server:arbiter-migrations:" <> schemaName :: Text) + + acquireArbiterMigrationLockWithTimeout :: HasqlConnection.Connection -> IO () + acquireArbiterMigrationLockWithTimeout connection = do + acquired <- timeout arbiterMigrationLockWaitTimeoutMicros retryUntilAcquired + case acquired of + Just () -> pure () + Nothing -> + throwIO . userError $ + "Timed out waiting for the Arbiter migration lock for schema " <> T.unpack schemaName + where + retryUntilAcquired :: IO () + retryUntilAcquired = do + acquired <- runAdvisoryLockStatement connection tryArbiterMigrationLock + if acquired + then pure () + else do + threadDelay arbiterMigrationLockRetryIntervalMicros + retryUntilAcquired + + -- Do not let a stuck migration block service startup indefinitely. The + -- migration itself is not subject to this timeout once the lock is held. + arbiterMigrationLockRetryIntervalMicros :: Int + arbiterMigrationLockRetryIntervalMicros = 1_000_000 + + arbiterMigrationLockWaitTimeoutMicros :: Int + arbiterMigrationLockWaitTimeoutMicros = 1 * 60 * 1_000_000 + + acquireConnection :: IO HasqlConnection.Connection + acquireConnection = do + connectionResult <- HasqlConnection.acquire . HasqlConnectionSettings.connectionString $ revealSecretText connStr + either + ( \err -> + throwIO . userError $ + "Failed to acquire PostgreSQL connection for Arbiter migration lock: " <> show err + ) + pure + connectionResult + + runAdvisoryLockStatement :: + HasqlConnection.Connection -> + HasqlStatement.Statement Int64 a -> + IO a + runAdvisoryLockStatement connection statement = do + result <- HasqlConnection.use connection (HasqlSession.statement lockId statement) + either + ( \err -> + throwIO . userError $ + "Arbiter migration advisory lock query failed: " <> show err + ) + pure + result + + tryArbiterMigrationLock :: HasqlStatement.Statement Int64 Bool + tryArbiterMigrationLock = + [singletonStatement|SELECT (pg_try_advisory_lock($1 :: bigint) :: bool)|] + + releaseArbiterMigrationLock :: HasqlStatement.Statement Int64 () + releaseArbiterMigrationLock = + [resultlessStatement|SELECT (1 :: integer) FROM (SELECT pg_advisory_unlock($1 :: bigint))|] + +-- This runner is specialized to the meetings cleanup payload for now. +-- If we add another cron job later, we can either reuse this helper with a +-- shared payload type or factor out the common Arbiter setup first. +runRecurringJobRunner :: + forall registry. + ( RegistryTables registry, + KnownSymbol (TableForPayload MeetingsCleanupJob registry), + FromJSON MeetingsCleanupJob, + ToJSON MeetingsCleanupJob + ) => + HasqlPoolExt.Pool -> + RecurringJobRunnerConfig registry -> + (MeetingsCleanupJob -> IO ()) -> + IO (IO ()) +runRecurringJobRunner postgresPool RecurringJobRunnerConfig {..} runJob = do + let arbiterConnStr = Text.encodeUtf8 (revealSecretText recurringJobRunnerArbiterConnStr) + Log.info recurringJobRunnerLogger $ + Log.msg (Log.val "Starting scheduled jobs worker") + . Log.field "job_name" recurringJobRunnerJobName + . Log.field "queue_name" recurringJobRunnerQueueName + . Log.field "schedule" (show recurringJobRunnerSchedule) + + let arbiterEnv = + WireArbiterEnv + { schemaName = recurringJobRunnerSchemaName, + connectionPool = postgresPool, + activeConn = Nothing, + transactionDepth = 0, + preparedStatements = False + } + + let workerHandler _conn job = + liftIO $ do + Log.info recurringJobRunnerLogger $ + Log.msg (Log.val "Running scheduled job") + . Log.field "job_name" recurringJobRunnerJobName + . Log.field "queue_name" recurringJobRunnerQueueName + runJob (ArbiterCore.payload job) + + cronJob = + case ArbiterWorkerCron.cronJob + recurringJobRunnerJobName + (serializeCronSchedule recurringJobRunnerSchedule) + ArbiterWorkerCron.SkipOverlap + ( \_ scheduledFor -> + (ArbiterCore.defaultGroupedJob recurringJobRunnerQueueName MeetingsCleanupJob) + { ArbiterCore.notVisibleUntil = Just scheduledFor, + ArbiterCore.maxAttempts = Just 3 + } + ) of + Left err -> error $ "Invalid cron schedule for " <> T.unpack recurringJobRunnerJobName <> ": " <> err + Right job -> job + + workerConfig <- + ( ArbiterWorker.defaultWorkerConfig + arbiterConnStr + recurringJobRunnerWorkerThreads + workerHandler :: + IO + ( ArbiterWorker.WorkerConfig + (WireArbiter registry) + MeetingsCleanupJob + () + ) + ) + let workerConfig' = + applyExplicitDefaults + recurringJobRunnerPollInterval + workerConfig + { ArbiterWorkerConfig.cronJobs = [cronJob] + } + + workerAsync <- + Async.async $ + runWireArbiter arbiterEnv $ + ArbiterWorker.runWorkerPool workerConfig' + + pure $ do + ArbiterWorker.shutdownWorker workerConfig' + Async.cancel workerAsync + +runOneOffJobRunner :: + forall registry (payload :: Type). + ( RegistryTables registry, + KnownSymbol (TableForPayload payload registry), + FromJSON payload, + ToJSON payload + ) => + HasqlPoolExt.Pool -> + OneOffJobRunnerConfig registry payload -> + (JobRead payload -> IO ()) -> + IO (IO ()) +runOneOffJobRunner postgresPool OneOffJobRunnerConfig {..} runJob = do + let arbiterConnStr = Text.encodeUtf8 (revealSecretText oneOffJobRunnerArbiterConnStr) + Log.info oneOffJobRunnerLogger $ + Log.msg (Log.val "Starting one-off jobs worker") + . Log.field "job_name" oneOffJobRunnerJobName + . Log.field "queue_name" oneOffJobRunnerQueueName + + let arbiterEnv = + WireArbiterEnv + { schemaName = oneOffJobRunnerSchemaName, + connectionPool = postgresPool, + activeConn = Nothing, + transactionDepth = 0, + preparedStatements = False + } + let workerHandler _conn job = + liftIO $ do + Log.info oneOffJobRunnerLogger $ + Log.msg (Log.val "Running one-off job") + . Log.field "job_name" oneOffJobRunnerJobName + . Log.field "queue_name" oneOffJobRunnerQueueName + runJob job + + workerConfig <- + ( ArbiterWorker.defaultWorkerConfig + arbiterConnStr + oneOffJobRunnerWorkerThreads + workerHandler :: + IO + ( ArbiterWorker.WorkerConfig + (WireArbiter registry) + payload + () + ) + ) + let workerConfig' = applyExplicitDefaults oneOffJobRunnerPollInterval workerConfig + + workerAsync <- + Async.async $ + runWireArbiter arbiterEnv $ + ArbiterWorker.runWorkerPool workerConfig' + + pure $ do + ArbiterWorker.shutdownWorker workerConfig' + Async.cancel workerAsync + +-- | Make the effective Arbiter defaults explicit in our code, even when we +-- keep the same values as 'defaultWorkerConfig'. +-- +-- This is intentionally redundant with Arbiter's own defaults. The point is to +-- show the runtime behavior in our repository and keep the knobs in one place +-- if we need to tune them later. +applyExplicitDefaults :: + NominalDiffTime -> + ArbiterWorker.WorkerConfig m payload result -> + ArbiterWorker.WorkerConfig m payload result +applyExplicitDefaults pollInterval cfg = + cfg + { -- How often the dispatcher wakes up to look for newly visible jobs. + -- Lower values reduce discovery latency at the cost of more DB traffic. + ArbiterWorkerConfig.pollInterval = pollInterval, + -- How long a claimed job stays invisible while a worker processes it. + -- Must exceed the job heartbeat interval so active jobs are not reclaimed. + ArbiterWorkerConfig.visibilityTimeout = 60, + -- How often a running job refreshes its visibility timeout. + -- Keeps long-running jobs from being reclaimed mid-flight. + ArbiterWorkerConfig.jobHeartbeatInterval = 30, + -- How often the worker process updates its own heartbeat and pause state. + -- This drives liveness, re-registration, and paused-state reconciliation. + ArbiterWorkerConfig.workerHeartbeatInterval = 10, + -- Retry strategy for transient worker failures. + -- Arbiter uses exponential backoff with jitter by default. + ArbiterWorkerConfig.backoffStrategy = ArbiterWorker.exponentialBackoff 2.0 1_048_576, + -- Jitter mode for retry delays. + -- Equal jitter smooths retry spikes without making them too aggressive. + ArbiterWorkerConfig.jitter = ArbiterWorker.EqualJitter, + -- How long the worker waits for in-flight jobs during shutdown. + -- If set, the pool exits after this grace period instead of waiting forever. + ArbiterWorkerConfig.gracefulShutdownTimeout = Just 30, + -- How often the reaper runs. It refreshes groups, sweeps stale workers, + -- and moves exhausted jobs to the DLQ. + ArbiterWorkerConfig.reaperInterval = 300, + -- How old a worker heartbeat may be before it is considered stale. + -- Stale workers are swept from the registry by the reaper. + ArbiterWorkerConfig.workerStaleThreshold = 300 + } diff --git a/libs/wire-subsystems/src/Wire/Options/Keys.hs b/libs/wire-subsystems/src/Wire/Options/Keys.hs index 53d3057ef78..ef25201119b 100644 --- a/libs/wire-subsystems/src/Wire/Options/Keys.hs +++ b/libs/wire-subsystems/src/Wire/Options/Keys.hs @@ -28,10 +28,6 @@ import Crypto.ECC hiding (KeyPair) import Crypto.Error import Crypto.PubKey.ECDSA qualified as ECDSA import Crypto.PubKey.Ed25519 qualified as Ed25519 -import Data.ASN1.BinaryEncoding -import Data.ASN1.BitArray -import Data.ASN1.Encoding -import Data.ASN1.Types import Data.Bifunctor import Data.ByteString.Lazy qualified as LBS import Data.PEM @@ -41,6 +37,10 @@ import Imports import Network.Wai.Utilities.Exception import Wire.API.MLS.CipherSuite import Wire.API.MLS.Keys +import "crypton-asn1-encoding" Data.ASN1.BinaryEncoding +import "crypton-asn1-encoding" Data.ASN1.Encoding +import "crypton-asn1-types" Data.ASN1.BitArray +import "crypton-asn1-types" Data.ASN1.Types type MLSPrivateKeyPaths = MLSKeysByPurpose (MLSKeys FilePath) diff --git a/libs/wire-subsystems/test/unit/Wire/ConversationSubsystem/InterpreterSpec.hs b/libs/wire-subsystems/test/unit/Wire/ConversationSubsystem/InterpreterSpec.hs index 34133bf4475..da60fb5045b 100644 --- a/libs/wire-subsystems/test/unit/Wire/ConversationSubsystem/InterpreterSpec.hs +++ b/libs/wire-subsystems/test/unit/Wire/ConversationSubsystem/InterpreterSpec.hs @@ -48,6 +48,7 @@ import Wire.ConversationSubsystem.Update (removeMemberQualified) import Wire.ExternalAccess (ExternalAccess (..)) import Wire.FeaturesConfigSubsystem (FeaturesConfigSubsystem (..)) import Wire.FederationAPIAccess (FederationAPIAccess (..)) +import Wire.JobSubsystem (JobSubsystem (..)) import Wire.MockInterpreters.Now (defaultTime, interpretNowConst) import Wire.MockInterpreters.TinyLog (noopLogger) import Wire.NotificationSubsystem (NotificationSubsystem (..)) @@ -121,6 +122,7 @@ spec = describe "ConversationSubsystem.Interpreter" do . interpretNotificationSubsystem . interpretProposalStore . interpretTeamSubsystem + . interpretJobSubsystem . interpretNowConst defaultTime . interpretRandom . noopLogger @@ -260,6 +262,17 @@ interpretTeamSubsystem = interpret $ \case _ -> error "unexpected TeamSubsystem call in test" +interpretJobSubsystem :: + Sem (JobSubsystem ': r) a -> + Sem r a +interpretJobSubsystem = + interpret $ \case + ScheduleAdminlessDeletionJob {} -> + pure () + ScheduleAdminlessReminderJob {} -> + pure () + StartJobWorkers _ _ -> pure (pure ()) + interpretRandom :: Sem (Random ': r) a -> Sem r a diff --git a/libs/wire-subsystems/wire-subsystems.cabal b/libs/wire-subsystems/wire-subsystems.cabal index 9ee955bcaf3..6f2ee5422ea 100644 --- a/libs/wire-subsystems/wire-subsystems.cabal +++ b/libs/wire-subsystems/wire-subsystems.cabal @@ -91,6 +91,12 @@ common common-all , amazonka-ses , amazonka-sqs , amqp + , arbiter-core + , arbiter-hasql + , arbiter-migrations + , arbiter-worker + , asn1-encoding + , asn1-types , async , attoparsec , base @@ -109,6 +115,7 @@ common common-all , contravariant , cookie , cql + , cron , crypton , crypton-asn1-encoding , crypton-asn1-types @@ -363,6 +370,10 @@ library Wire.InternalEvent Wire.InvitationStore Wire.InvitationStore.Cassandra + Wire.JobSubsystem + Wire.JobSubsystem.ArbiterAdapter + Wire.JobSubsystem.Interpreter + Wire.JobSubsystem.Workers Wire.LegalHold Wire.LegalHoldStore Wire.LegalHoldStore.Cassandra diff --git a/nix/haskell-pins.nix b/nix/haskell-pins.nix index 17849cbc878..07b0f2c3fa3 100644 --- a/nix/haskell-pins.nix +++ b/nix/haskell-pins.nix @@ -58,6 +58,18 @@ let # END maintained by us # -------------------- + arbiter = { + src = inputs.arbiter; + packages = { + arbiter-core = "arbiter-core"; + arbiter-hasql = "arbiter-hasql"; + arbiter-migrations = "arbiter-migrations"; + arbiter-simple = "arbiter-simple"; + arbiter-test-common = "arbiter-test-common"; + arbiter-worker = "arbiter-worker"; + }; + }; + bloodhound = { src = inputs.bloodhound; }; diff --git a/nix/manual-overrides.nix b/nix/manual-overrides.nix index 59373e631c3..e1096aec75b 100644 --- a/nix/manual-overrides.nix +++ b/nix/manual-overrides.nix @@ -32,6 +32,12 @@ hself: hsuper: { hasql-migration = hlib.markUnbroken (hlib.doJailbreak (hlib.dontCheck hsuper.hasql-migration)); hasql-transaction = hlib.dontCheck hsuper.hasql-transaction_1_2_2; postgresql-binary = hlib.dontCheck (hsuper.postgresql-binary_0_15_0_1); + monad-logger-aeson = hlib.markUnbroken (hlib.dontCheck hsuper.monad-logger-aeson); + arbiter-core = hlib.markUnbroken (hlib.dontCheck hsuper.arbiter-core); + arbiter-hasql = hlib.markUnbroken (hlib.dontCheck hsuper.arbiter-hasql); + arbiter-migrations = hlib.markUnbroken (hlib.dontCheck hsuper.arbiter-migrations); + arbiter-simple = hlib.markUnbroken (hlib.dontCheck hsuper.arbiter-simple); + arbiter-worker = hlib.markUnbroken (hlib.dontCheck hsuper.arbiter-worker); # Test fixtures don't seem to be bundled for Hackage hsaml2 = hlib.dontCheck (hsuper.hsaml2); diff --git a/services/background-worker/background-worker.cabal b/services/background-worker/background-worker.cabal index af6b66a8ffe..ffaadd8cfbd 100644 --- a/services/background-worker/background-worker.cabal +++ b/services/background-worker/background-worker.cabal @@ -12,6 +12,7 @@ build-type: Simple library -- cabal-fmt: expand src exposed-modules: + Wire.AdminlessJobsWorker Wire.BackendNotificationPusher Wire.BackgroundWorker Wire.BackgroundWorker.Env @@ -19,6 +20,7 @@ library Wire.BackgroundWorker.Jobs.Consumer Wire.BackgroundWorker.Jobs.Registry Wire.BackgroundWorker.Options + Wire.BackgroundWorker.ScheduledJobs Wire.BackgroundWorker.Util Wire.DeadUserNotificationWatcher Wire.Effects @@ -35,6 +37,7 @@ library build-depends: aeson , amqp + , arbiter-core , base , bilge , bytestring diff --git a/services/background-worker/background-worker.integration.yaml b/services/background-worker/background-worker.integration.yaml index 97734fccd3b..7491379215f 100644 --- a/services/background-worker/background-worker.integration.yaml +++ b/services/background-worker/background-worker.integration.yaml @@ -65,6 +65,10 @@ backgroundJobs: jobTimeout: 5s maxAttempts: 3 +# Scheduled-jobs dispatcher configuration for integration +scheduledJobs: + pollInterval: 1s # Poll every second so due jobs are discovered promptly in tests + # Meetings cleanup configuration for integration meetingsCleanup: cleanOlderThanHours: 0.0014 # Clean meetings older than ~5 seconds diff --git a/services/background-worker/default.nix b/services/background-worker/default.nix index 28faab430e9..7a07604b93f 100644 --- a/services/background-worker/default.nix +++ b/services/background-worker/default.nix @@ -5,6 +5,7 @@ { mkDerivation , aeson , amqp +, arbiter-core , base , bilge , bytestring @@ -66,6 +67,7 @@ mkDerivation { libraryHaskellDepends = [ aeson amqp + arbiter-core base bilge bytestring diff --git a/services/background-worker/src/Wire/AdminlessJobsWorker.hs b/services/background-worker/src/Wire/AdminlessJobsWorker.hs new file mode 100644 index 00000000000..2cc05d9bb4d --- /dev/null +++ b/services/background-worker/src/Wire/AdminlessJobsWorker.hs @@ -0,0 +1,83 @@ +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or +-- FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License +-- for more details. +-- +-- You should have received a copy of the GNU Affero General Public License +-- along with this program. If not, see . + +module Wire.AdminlessJobsWorker + ( runAdminlessDeletionJob, + runAdminlessReminderJob, + ) +where + +import Arbiter.Core.Job.Types (JobRead, notVisibleUntil, payload) +import Data.Id (RequestId (..)) +import Data.Qualified (toLocalUnsafe) +import Imports +import System.Logger qualified as Log +import Wire.API.Jobs (AdminlessDeletionJob (..), AdminlessReminderJob (..)) +import Wire.BackgroundWorker.Env (AppT, Env (..)) +import Wire.ConversationSubsystem +import Wire.Effects (runBackgroundWorkerEffects) +import Wire.ExternalAccess.External (ExtEnv) + +runAdminlessDeletionJob :: ExtEnv -> JobRead AdminlessDeletionJob -> AppT IO () +runAdminlessDeletionJob extEnv job = do + env <- ask + let jobPayload = payload job + Log.info env.logger $ + Log.msg (Log.val "Running adminless deletion job") + . Log.field "team_id" (show (adminlessDeletionJobTeamId jobPayload)) + . Log.field "conversation_id" (show (adminlessDeletionJobConversationId jobPayload)) + . Log.field "orig_user_id" (show (adminlessDeletionJobOrigUserId jobPayload)) + . Log.field "scheduled_for" (show (notVisibleUntil job)) + result <- + liftIO $ + runBackgroundWorkerEffects env extEnv (RequestId "adminless-deletion") Nothing $ + do + Log.info env.logger $ + Log.msg (Log.val "Adminless deletion job: invoking conversation delete") + . Log.field "team_id" (show (adminlessDeletionJobTeamId jobPayload)) + . Log.field "conversation_id" (show (adminlessDeletionJobConversationId jobPayload)) + internalDeleteLocalAdminlessGroup + (toLocalUnsafe env.federationDomain (adminlessDeletionJobOrigUserId jobPayload)) + (toLocalUnsafe env.federationDomain (adminlessDeletionJobConversationId jobPayload)) + Log.info env.logger $ + Log.msg (Log.val "Adminless deletion job finished") + . Log.field "team_id" (show (adminlessDeletionJobTeamId jobPayload)) + . Log.field "conversation_id" (show (adminlessDeletionJobConversationId jobPayload)) + either (liftIO . fail . show) pure result + +runAdminlessReminderJob :: ExtEnv -> JobRead AdminlessReminderJob -> AppT IO () +runAdminlessReminderJob extEnv job = do + env <- ask + let jobPayload = payload job + Log.info env.logger $ + Log.msg (Log.val "Running adminless reminder job") + . Log.field "team_id" (show (adminlessReminderJobTeamId jobPayload)) + . Log.field "conversation_id" (show (adminlessReminderJobConversationId jobPayload)) + . Log.field "deletion_scheduled_for" (show (adminlessReminderJobDeletionScheduledFor jobPayload)) + result <- + liftIO $ + runBackgroundWorkerEffects env extEnv (RequestId "adminless-reminder") Nothing $ + do + internalNotifyAdminlessReminder + (toLocalUnsafe env.federationDomain (adminlessReminderJobOrigUserId jobPayload)) + (toLocalUnsafe env.federationDomain (adminlessReminderJobConversationId jobPayload)) + (adminlessReminderJobDeletionScheduledFor jobPayload) + Log.info env.logger $ + Log.msg (Log.val "Adminless reminder job finished") + . Log.field "team_id" (show (adminlessReminderJobTeamId jobPayload)) + . Log.field "conversation_id" (show (adminlessReminderJobConversationId jobPayload)) + either (liftIO . fail . show) pure result diff --git a/services/background-worker/src/Wire/BackgroundWorker.hs b/services/background-worker/src/Wire/BackgroundWorker.hs index fd145cf4668..ffee3ff23bf 100644 --- a/services/background-worker/src/Wire/BackgroundWorker.hs +++ b/services/background-worker/src/Wire/BackgroundWorker.hs @@ -33,8 +33,8 @@ import Wire.BackgroundWorker.Env import Wire.BackgroundWorker.Health qualified as Health import Wire.BackgroundWorker.Jobs.Consumer qualified as Jobs import Wire.BackgroundWorker.Options +import Wire.BackgroundWorker.ScheduledJobs qualified as ScheduledJobs import Wire.DeadUserNotificationWatcher qualified as DeadUserNotificationWatcher -import Wire.MeetingsCleanupWorker qualified as MeetingsCleanupWorker import Wire.Options.Galley qualified as Galley import Wire.PostgresMigrations qualified as Migrations @@ -82,10 +82,10 @@ run opts galleyOpts = do runAppT env $ withNamedLogger "background-job-consumer" $ Jobs.startWorker amqpEP - cleanupMeetings <- + cleanupScheduledJobs <- runAppT env $ - withNamedLogger "meetings-cleanup" $ - MeetingsCleanupWorker.startWorker opts.meetingsCleanup + withNamedLogger "scheduled-jobs" $ + ScheduledJobs.startWorker opts.scheduledJobs opts.meetingsCleanup let cleanup = void $ runConcurrently $ @@ -97,7 +97,7 @@ run opts galleyOpts = do <*> Concurrently cleanupTeamFeaturesMigration <*> Concurrently cleanupDomainRegistrationMigration <*> Concurrently cleanupJobs - <*> Concurrently cleanupMeetings + <*> Concurrently cleanupScheduledJobs let server = defaultServer (T.unpack opts.backgroundWorker.host) opts.backgroundWorker.port env.logger let settings = newSettings server diff --git a/services/background-worker/src/Wire/BackgroundWorker/Env.hs b/services/background-worker/src/Wire/BackgroundWorker/Env.hs index 2bccadee873..bc904921a4f 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Env.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Env.hs @@ -29,6 +29,7 @@ import Data.Domain (Domain) import Data.Id (TeamId) import Data.Map.Strict qualified as Map import Data.Misc (HttpsUrl) +import Data.Secret (SecretText) import HTTP2.Client.Manager import Hasql.Pool.Extended import Hasql.Pool.Extended qualified as Hasql @@ -89,6 +90,8 @@ data Env = Env cassandraGalley :: ClientState, cassandraBrig :: ClientState, hasqlPool :: Hasql.Pool, + -- May contain the PostgreSQL password. Do not unwrap outside the Arbiter boundary. + arbiterConnStr :: SecretText, -- Dedicated AMQP channels per concern amqpJobsPublisherChannel :: MVar Q.Channel, amqpBackendNotificationsChannel :: MVar Q.Channel, @@ -190,6 +193,7 @@ mkEnv opts galleyOpts = do checkGroupInfo = galleyOpts._settings._checkGroupInfo workerRunningGauge <- mkWorkerRunningGauge hasqlPool <- initPostgresPool opts.postgresqlPool galleyOpts._postgresql galleyOpts._postgresqlPassword + arbiterConnStr <- postgresqlConnectionStringWithPassword galleyOpts._postgresql galleyOpts._postgresqlPassword Log.info logger $ Log.msg @Text "Opening RabbitMQ channel: background-worker-jobs-publisher..." amqpJobsPublisherChannel <- mkRabbitMqChannelMVar logger (Just "background-worker-jobs-publisher") $ diff --git a/services/background-worker/src/Wire/BackgroundWorker/Options.hs b/services/background-worker/src/Wire/BackgroundWorker/Options.hs index f89c52e8300..e6e90c293a3 100644 --- a/services/background-worker/src/Wire/BackgroundWorker/Options.hs +++ b/services/background-worker/src/Wire/BackgroundWorker/Options.hs @@ -55,6 +55,7 @@ data Opts = Opts migrateConversationCodes :: !Bool, migrateTeamFeatures :: !Bool, migrateDomainRegistration :: !Bool, + scheduledJobs :: ScheduledJobsConfig, meetingsCleanup :: MeetingsCleanupConfig, backgroundJobs :: BackgroundJobsConfig } @@ -99,6 +100,22 @@ data BackgroundJobsConfig = BackgroundJobsConfig deriving (Show, Generic) deriving (FromJSON) via Generically BackgroundJobsConfig +data ScheduledJobsConfig = ScheduledJobsConfig + { -- | Arbiter dispatcher poll interval for scheduled jobs. + -- Lower values reduce discovery latency for due jobs. + pollInterval :: Duration + } + deriving (Show, Generic) + +instance FromJSON ScheduledJobsConfig where + parseJSON = + withObject "ScheduledJobsConfig" $ \o -> do + pollInterval <- o .: "pollInterval" + when (duration pollInterval <= 0) $ + parserThrowError [Key "pollInterval"] $ + "pollInterval must be greater than 0, got: " <> show pollInterval + pure ScheduledJobsConfig {..} + data MeetingsCleanupConfig = MeetingsCleanupConfig { -- | Delete meetings older than this many hours cleanOlderThanHours :: Double, diff --git a/services/background-worker/src/Wire/BackgroundWorker/ScheduledJobs.hs b/services/background-worker/src/Wire/BackgroundWorker/ScheduledJobs.hs new file mode 100644 index 00000000000..1d625c9f5db --- /dev/null +++ b/services/background-worker/src/Wire/BackgroundWorker/ScheduledJobs.hs @@ -0,0 +1,115 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE TypeFamilies #-} + +-- This file is part of the Wire Server implementation. +-- +-- Copyright (C) 2026 Wire Swiss GmbH +-- +-- This program is free software: you can redistribute it and/or modify it under +-- the terms of the GNU Affero General Public License as published by the Free +-- Software Foundation, either version 3 of the License, or (at your option) +-- any later version. +-- +-- This program is distributed in the hope that it will be useful, but WITHOUT +-- ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS +-- FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more +-- details. +-- +-- You should have received a copy of the GNU Affero General Public License along +-- with this program. If not, see . + +module Wire.BackgroundWorker.ScheduledJobs (startWorker) where + +import Arbiter.Core qualified as ArbiterCore +import Data.Id (RequestId (..)) +import Data.Misc (Duration, durationToMicros) +import Data.Time.Clock (NominalDiffTime) +import Imports +import Wire.API.Jobs +import Wire.AdminlessJobsWorker (runAdminlessDeletionJob, runAdminlessReminderJob) +import Wire.BackgroundWorker.Env (AppT, Env (..), runAppT) +import Wire.BackgroundWorker.Options (MeetingsCleanupConfig (..), ScheduledJobsConfig (..)) +import Wire.Effects (runBackgroundWorkerEffects) +import Wire.ExternalAccess.External (initExtEnv) +import Wire.JobSubsystem +import Wire.JobSubsystem.Workers +import Wire.MeetingsCleanupWorker + +-- | Initial worker pool size for the scheduled-jobs queues. +-- +-- Keep this explicit so the next tuning pass has a single obvious place to +-- change the parallelism for meetings cleanup and adminless jobs. +scheduledJobsWorkerThreads :: Int +scheduledJobsWorkerThreads = 1 + +startWorker :: ScheduledJobsConfig -> MeetingsCleanupConfig -> AppT IO CleanupAction +startWorker scheduledConfig config = do + env <- ask + extEnv <- liftIO $ initExtEnv True + let cleanupConfig = + CleanupConfig + { retentionHours = config.cleanOlderThanHours, + batchSize = config.batchSize + } + jobHandlers = + JobWorkerHandlers + { recurringJobRunnerRunJob = \_ -> + runAppT env $ + runCleanupOldMeetings cleanupConfig, + adminlessDeletionJobRunnerRunJob = \job -> + runAppT env $ + runAdminlessDeletionJob extEnv job, + adminlessReminderJobRunnerRunJob = \job -> + runAppT env $ + runAdminlessReminderJob extEnv job + } + workersConfig = + JobWorkersConfig + { recurringJobRunnerConfig = + RecurringJobRunnerConfig + { recurringJobRunnerLogger = env.logger, + recurringJobRunnerSchedule = config.schedule, + -- Arbiter still uses the connection string for LISTEN/NOTIFY. + -- The actual job DB access goes through the shared Hasql pool + -- pulled from the JobSubsystem interpreter. + recurringJobRunnerArbiterConnStr = env.arbiterConnStr, + recurringJobRunnerSchemaName = ArbiterCore.defaultSchemaName, + recurringJobRunnerPollInterval = scheduledJobsPollIntervalSeconds scheduledConfig.pollInterval, + recurringJobRunnerWorkerThreads = scheduledJobsWorkerThreads, + recurringJobRunnerJobName = "meetings-cleanup", + recurringJobRunnerQueueName = meetingsCleanupQueueName + }, + adminlessDeletionJobRunnerConfig = + OneOffJobRunnerConfig + { oneOffJobRunnerLogger = env.logger, + -- Arbiter still uses the connection string for LISTEN/NOTIFY. + -- The actual job DB access goes through the shared Hasql pool + -- pulled from the JobSubsystem interpreter. + oneOffJobRunnerArbiterConnStr = env.arbiterConnStr, + oneOffJobRunnerSchemaName = ArbiterCore.defaultSchemaName, + oneOffJobRunnerPollInterval = scheduledJobsPollIntervalSeconds scheduledConfig.pollInterval, + oneOffJobRunnerWorkerThreads = scheduledJobsWorkerThreads, + oneOffJobRunnerJobName = "adminless-deletion", + oneOffJobRunnerQueueName = adminlessDeletionQueueName + }, + adminlessReminderJobRunnerConfig = + OneOffJobRunnerConfig + { oneOffJobRunnerLogger = env.logger, + oneOffJobRunnerArbiterConnStr = env.arbiterConnStr, + oneOffJobRunnerSchemaName = ArbiterCore.defaultSchemaName, + oneOffJobRunnerPollInterval = scheduledJobsPollIntervalSeconds scheduledConfig.pollInterval, + oneOffJobRunnerWorkerThreads = scheduledJobsWorkerThreads, + oneOffJobRunnerJobName = "adminless-reminder", + oneOffJobRunnerQueueName = adminlessReminderQueueName + } + } + result <- + liftIO $ + runBackgroundWorkerEffects env extEnv (RequestId "scheduled-jobs") Nothing $ + startJobWorkers workersConfig jobHandlers + either (liftIO . fail . show) pure result + +-- | Convert the explicit background-worker config into the Arbiter poll +-- interval in seconds. +scheduledJobsPollIntervalSeconds :: Duration -> NominalDiffTime +scheduledJobsPollIntervalSeconds = (/ 1_000_000) . fromIntegral . durationToMicros diff --git a/services/background-worker/src/Wire/Effects.hs b/services/background-worker/src/Wire/Effects.hs index 57330f9f1e3..947a03f25af 100644 --- a/services/background-worker/src/Wire/Effects.hs +++ b/services/background-worker/src/Wire/Effects.hs @@ -20,6 +20,7 @@ module Wire.Effects ) where +import Arbiter.Core qualified as ArbiterCore import Bilge qualified import Bilge.Retry import Cassandra (ClientState) @@ -90,9 +91,12 @@ import Wire.GalleyAPIAccess.Rpc (interpretGalleyAPIAccessToRpc) import Wire.GundeckAPIAccess import Wire.HashPassword (HashPassword) import Wire.HashPassword.Interpreter (runHashPassword) +import Wire.JobSubsystem (JobSubsystem, JobSubsystemConfig (..)) +import Wire.JobSubsystem.Interpreter (interpretJobSubsystem) import Wire.LegalHoldStore (LegalHoldStore) import Wire.LegalHoldStore.Cassandra (interpretLegalHoldStoreToCassandra) import Wire.LegalHoldStore.Env (LegalHoldEnv (..)) +import Wire.MigrationLock (MigrationLockError) import Wire.NotificationSubsystem (NotificationSubsystem) import Wire.NotificationSubsystem.Interpreter import Wire.Options.Galley (GuestLinkTTLSeconds) @@ -124,6 +128,8 @@ import Wire.TeamCollaboratorsSubsystem.Interpreter (interpretTeamCollaboratorsSu import Wire.TeamFeatureStore (TeamFeatureStore) import Wire.TeamFeatureStore.Cassandra (interpretTeamFeatureStoreToCassandra) import Wire.TeamFeatureStore.Error (TeamFeatureStoreError) +import Wire.TeamFeatureStore.Migrating (interpretTeamFeatureStoreToCassandraAndPostgres) +import Wire.TeamFeatureStore.Postgres (interpretTeamFeatureStoreToPostgres) import Wire.TeamJournal (TeamJournal) import Wire.TeamJournal.Aws (interpretTeamJournal) import Wire.TeamStore (TeamStore) @@ -210,6 +216,7 @@ type BackgroundWorkerEffects = Now, TeamJournal, LegalHoldStore, + JobSubsystem, TeamCollaboratorsStore, TeamStore, ConversationStore, @@ -242,6 +249,7 @@ type BackgroundWorkerEffects = Error (Tagged CodeStoreNotFound ()), Error TeamFeatureStoreError, Error TeamCollaboratorsError, + Error MigrationLockError, Error UnreachableBackends, Error InternalError, Error MigrationError, @@ -290,6 +298,7 @@ runBackgroundWorkerEffects env extEnv requestId mJobId = . mapError @MigrationError (T.pack . show) . mapError @InternalError (TL.toStrict . internalErrorDescription) . mapError @UnreachableBackends (T.pack . show) + . mapError @MigrationLockError (const ("Migration lock error" :: Text)) . mapError @TeamCollaboratorsError (const ("Team collaborators error" :: Text)) . mapError @TeamFeatureStoreError (const ("Team feature store error" :: Text)) . mapError @(Tagged 'CodeStoreNotFound ()) (const ("Code store not found" :: Text)) @@ -317,11 +326,12 @@ runBackgroundWorkerEffects env extEnv requestId mJobId = . interpretProposalStoreToCassandra . interpretServiceStoreToCassandra env.cassandraBrig . interpretUserGroupStoreToPostgres - . interpretTeamFeatureStoreToCassandra + . interpretTeamFeatureStore . interpretUserClientIndexStoreToCassandra env.cassandraGalley . interpretConversationStoreByMigration env.postgresMigration.conversation env.cassandraGalley . interpretTeamStoreToCassandra . interpretTeamCollaboratorsStoreToPostgres + . interpretJobSubsystem jobSubsystemConfig . interpretLegalHoldStoreToCassandra (env.conversationSubsystemConfig.legalholdDefaults) . interpretTeamJournal Nothing . nowToIO @@ -355,11 +365,15 @@ runBackgroundWorkerEffects env extEnv requestId mJobId = . interpretTeamCollaboratorsSubsystem . interpretConversationSubsystem where - convCodesStoreInterpreter = - case env.postgresMigration.conversationCodes of - CassandraStorage -> interpretCodeStoreToCassandra - MigrationToPostgresql -> interpretCodeStoreToCassandraAndPostgres - PostgresqlStorage -> interpretCodeStoreToPostgres + interpretTeamFeatureStore = case env.postgresMigration.teamFeatures of + CassandraStorage -> interpretTeamFeatureStoreToCassandra + MigrationToPostgresql -> interpretTeamFeatureStoreToCassandraAndPostgres + PostgresqlStorage -> interpretTeamFeatureStoreToPostgres + + convCodesStoreInterpreter = case env.postgresMigration.conversationCodes of + CassandraStorage -> interpretCodeStoreToCassandra + MigrationToPostgresql -> interpretCodeStoreToCassandraAndPostgres + PostgresqlStorage -> interpretCodeStoreToPostgres legalHoldEnv = let makeReq fpr url rb = makeVerifiedRequestIO env.logger extEnv fpr url rb makeReqFresh fpr url rb = makeVerifiedRequestFreshManagerIO env.logger fpr url rb @@ -372,6 +386,10 @@ runBackgroundWorkerEffects env extEnv requestId mJobId = http2Manager = env.http2Manager, requestId = requestId } + jobSubsystemConfig = + JobSubsystemConfig + { jobSubsystemSchemaName = ArbiterCore.defaultSchemaName + } backendQueueEnv = BackendNotificationQueueAccess.Env { channelMVar = env.amqpBackendNotificationsChannel, diff --git a/services/background-worker/src/Wire/MeetingsCleanupWorker.hs b/services/background-worker/src/Wire/MeetingsCleanupWorker.hs index 1fd4db93ff9..5b00955373f 100644 --- a/services/background-worker/src/Wire/MeetingsCleanupWorker.hs +++ b/services/background-worker/src/Wire/MeetingsCleanupWorker.hs @@ -16,8 +16,8 @@ -- with this program. If not, see . module Wire.MeetingsCleanupWorker - ( startWorker, - CleanupConfig (..), + ( CleanupConfig (..), + runCleanupOldMeetings, ) where @@ -28,11 +28,8 @@ import Data.Time.Clock import Imports import Polysemy.Error (runError) import Prometheus (incCounter) -import System.Cron (Job (..), forkJob) import System.Logger qualified as Log -import Wire.BackgroundWorker.Env (AppT, Env (..), MeetingsCleanupMetrics (..), runAppT) -import Wire.BackgroundWorker.Options (MeetingsCleanupConfig (..)) -import Wire.BackgroundWorker.Util (CleanupAction) +import Wire.BackgroundWorker.Env (AppT, Env (..), MeetingsCleanupMetrics (..)) import Wire.Effects import Wire.ExternalAccess.External import Wire.MeetingsStore.Postgres (interpretMeetingsStoreToPostgres) @@ -45,37 +42,6 @@ data CleanupConfig = CleanupConfig } deriving (Show, Eq) --- | Start the meetings cleanup worker thread --- --- This worker runs periodically to clean up old meetings based on the configuration. -startWorker :: - MeetingsCleanupConfig -> - AppT IO CleanupAction -startWorker config = do - env <- ask - Log.info env.logger $ - Log.msg (Log.val "Starting meetings cleanup worker") - . Log.field "schedule" (show config.schedule) - . Log.field "clean_older_than_hours" config.cleanOlderThanHours - - void . liftIO $ do - forkJob $ - Job config.schedule $ - runAppT env $ do - Log.info env.logger $ Log.msg (Log.val "Starting scheduled meetings cleanup") - runCleanupOldMeetings (configFromOptions config) - liftIO $ incCounter env.meetingsCleanupMetrics.runsCounter - - pure $ pure () - --- | Convert MeetingsCleanupConfig to CleanupConfig -configFromOptions :: MeetingsCleanupConfig -> CleanupConfig -configFromOptions cfg = - CleanupConfig - { retentionHours = cfg.cleanOlderThanHours, - batchSize = cfg.batchSize - } - -- | Main cleanup function that orchestrates the cleanup process runCleanupOldMeetings :: CleanupConfig -> AppT IO () runCleanupOldMeetings config = do @@ -95,6 +61,7 @@ runCleanupOldMeetings config = do Log.info env.logger $ Log.msg (Log.val "Completed cleanup of old meetings") . Log.field "total_deleted" totalDeleted + liftIO $ incCounter env.meetingsCleanupMetrics.runsCounter cleanupLoop :: Env -> UTCTime -> NominalDiffTime -> Int -> Int64 -> AppT IO Int64 cleanupLoop env cutoffTime validityPeriod batchSize totalSoFar = do diff --git a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs index 4b32861ba3e..7222120d93a 100644 --- a/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs +++ b/services/background-worker/test/Test/Wire/BackendNotificationPusherSpec.hs @@ -32,6 +32,7 @@ import Data.Domain import Data.Id import Data.Misc import Data.Range +import Data.Secret (secretText) import Data.Sequence qualified as Seq import Data.Text qualified as Text import Data.Text.Encoding qualified as Text @@ -384,6 +385,7 @@ spec = do guestLinkTTLSeconds = Nothing passwordHashingOptions = PasswordHashingScrypt checkGroupInfo = Nothing + arbiterConnStr = secretText "" convCodeURI = Left (fromRight (error "Failed to parse test HttpsUrl") $ httpsUrlFromText "https://localhost") featureFlags = def conversationSubsystemConfig = @@ -447,6 +449,7 @@ spec = do guestLinkTTLSeconds = Nothing passwordHashingOptions = PasswordHashingScrypt checkGroupInfo = Nothing + arbiterConnStr = secretText "" convCodeURI = Left (fromRight (error "Failed to parse test HttpsUrl") $ httpsUrlFromText "https://localhost") featureFlags = def conversationSubsystemConfig = diff --git a/services/background-worker/test/Test/Wire/Util.hs b/services/background-worker/test/Test/Wire/Util.hs index ddecd07b9e9..5d89532bfec 100644 --- a/services/background-worker/test/Test/Wire/Util.hs +++ b/services/background-worker/test/Test/Wire/Util.hs @@ -24,6 +24,7 @@ import Data.Domain (Domain (Domain)) import Data.Misc import Data.Proxy import Data.Range +import Data.Secret (secretText) import Imports import Network.HTTP.Client hiding (Proxy) import System.Logger.Class qualified as Logger @@ -83,6 +84,7 @@ testEnv = do guestLinkTTLSeconds = Nothing passwordHashingOptions = PasswordHashingScrypt checkGroupInfo = Nothing + arbiterConnStr = secretText "" convCodeURI = Left (fromRight (error "Failed to parse test HttpsUrl") $ httpsUrlFromText "https://localhost") featureFlags = def conversationSubsystemConfig = diff --git a/services/brig/src/Brig/App.hs b/services/brig/src/Brig/App.hs index 9a4a59dda72..dd1192ccf3f 100644 --- a/services/brig/src/Brig/App.hs +++ b/services/brig/src/Brig/App.hs @@ -137,7 +137,7 @@ import Data.Text.IO qualified as Text import Data.Time.Clock import Database.Bloodhound qualified as ES import HTTP2.Client.Manager (Http2Manager, http2ManagerWithSSLCtx) -import Hasql.Pool.Extended +import Hasql.Pool.Extended (initPostgresPool) import Hasql.Pool.Extended qualified as HasqlPool import Imports import Network.AMQP qualified as Q diff --git a/services/galley/default.nix b/services/galley/default.nix index 3f9eb5cb466..949e941b3a9 100644 --- a/services/galley/default.nix +++ b/services/galley/default.nix @@ -7,6 +7,7 @@ , aeson-qq , amazonka , amqp +, arbiter-core , async , base , base64-bytestring @@ -121,6 +122,7 @@ mkDerivation { aeson amazonka amqp + arbiter-core async base bilge diff --git a/services/galley/galley.cabal b/services/galley/galley.cabal index 5ca4fdc34c8..f7c48e3bc00 100644 --- a/services/galley/galley.cabal +++ b/services/galley/galley.cabal @@ -197,6 +197,7 @@ library , aeson >=2.0.1.0 , amazonka >=1.4.5 , amqp + , arbiter-core , async >=2.0 , base >=4.6 && <5 , bilge >=0.21.1 diff --git a/services/galley/src/Galley/App.hs b/services/galley/src/Galley/App.hs index 78183816268..778168d3c37 100644 --- a/services/galley/src/Galley/App.hs +++ b/services/galley/src/Galley/App.hs @@ -42,6 +42,7 @@ module Galley.App ) where +import Arbiter.Core qualified as ArbiterCore import Bilge hiding (Request, header, host, options, port, statusCode, statusMessage) import Cassandra hiding (Set) import Cassandra.Util (initCassandraForService) @@ -124,6 +125,8 @@ import Wire.FireAndForget import Wire.GundeckAPIAccess (GundeckAPIAccess, runGundeckAPIAccess) import Wire.HashPassword import Wire.HashPassword.Interpreter +import Wire.JobSubsystem (JobSubsystem, JobSubsystemConfig (..)) +import Wire.JobSubsystem.Interpreter (interpretJobSubsystem) import Wire.LegalHoldStore (LegalHoldStore) import Wire.LegalHoldStore.Cassandra (interpretLegalHoldStoreToCassandra) import Wire.LegalHoldStore.Env (LegalHoldEnv (..)) @@ -191,6 +194,7 @@ import Wire.UserGroupStore.Postgres (interpretUserGroupStoreToPostgres) type GalleyEffects = '[ MeetingsSubsystem, ConversationSubsystem, + JobSubsystem, FederationSubsystem, TeamCollaboratorsSubsystem, Input AllTeamFeatures, @@ -549,6 +553,10 @@ evalGalley e = . runInputSem getAllTeamFeaturesForServer . interpretTeamCollaboratorsSubsystem . runFederationSubsystem conversationSubsystemConfig.federationProtocols + . interpretJobSubsystem + JobSubsystemConfig + { jobSubsystemSchemaName = ArbiterCore.defaultSchemaName + } . interpretConversationSubsystem . Meeting.interpretMeetingsSubsystem meetingValidityPeriod where diff --git a/services/galley/src/Galley/Run.hs b/services/galley/src/Galley/Run.hs index 9ea6b3870ae..67a3b316f33 100644 --- a/services/galley/src/Galley/Run.hs +++ b/services/galley/src/Galley/Run.hs @@ -23,6 +23,7 @@ where import AWS.Util (readAuthExpiration) import Amazonka qualified as AWS +import Arbiter.Core qualified as ArbiterCore import Cassandra (runClient, shutdown) import Cassandra.Schema (versionCheck) import Control.Concurrent.Async qualified as Async @@ -45,7 +46,7 @@ import Galley.Cassandra import Galley.Env import Galley.Monad import Galley.Queue qualified as Q -import Hasql.Pool.Extended (rawPool) +import Hasql.Pool.Extended (postgresqlConnectionStringWithPassword, rawPool) import Imports import Network.HTTP.Media.RenderHeader qualified as HTTPMedia import Network.HTTP.Types qualified as HTTP @@ -67,6 +68,7 @@ import Wire.API.Routes.Public.Galley import Wire.API.Routes.Version import Wire.API.Routes.Version.Wai import Wire.AWS (awsEnv) +import Wire.JobSubsystem.Workers (runScheduledJobsMigrations) import Wire.OpenTelemetry (withTracerC) import Wire.Options.Galley import Wire.PostgresMigrations (runAllMigrations) @@ -76,6 +78,12 @@ run opts = lowerCodensity do tracer <- withTracerC (app, env) <- mkApp opts lift $ runAllMigrations env._hasqlPool.rawPool env._applog + arbiterConnStr <- + lift $ + postgresqlConnectionStringWithPassword + (opts ^. postgresql) + (opts ^. postgresqlPassword) + lift $ runScheduledJobsMigrations arbiterConnStr ArbiterCore.defaultSchemaName let settings' = newSettings $ defaultServer