WPB-24076: Add meeting cleaner job in background-worker

Author: blackheaven

charts/wire-server/templates/background-worker/configmap.yaml

@@ -90,6 +90,10 @@ data:
 {{toYaml .backendNotificationPusher | indent 6 }}
     {{- with .backgroundJobs }}
     backgroundJobs:
+{{ toYaml . | indent 6 }}
+    {{- end }}
+    {{- with .meetingsCleanup }}
+    meetingsCleanup:
 {{ toYaml . | indent 6 }}
     {{- end }}
     {{- if .postgresMigration }}

integration/test/Test/Meetings.hs

@@ -4,10 +4,16 @@ module Test.Meetings where
 
 import API.Galley
 import qualified API.GalleyInternal as I
+import Control.Monad.Reader (ask)
+import qualified Data.Text as Text
+import qualified Data.Text.Encoding as Text
 import Data.Time.Clock
 import qualified Data.Time.Format as Time
 import SetupHelpers
+import System.Timeout (timeout)
 import Testlib.Prelude
+import Text.Regex.TDFA ((=~))
+import UnliftIO.Concurrent (threadDelay)
 
 -- Helper to extract meetingId and domain from a meeting JSON object
 getMeetingIdAndDomain :: (HasCallStack) => Value -> App (String, String)
@@ -311,3 +317,93 @@ testMeetingRemoveInvitationNotFound = do
   let removeInvitation = object ["emails" .= ["alice@example.com"]]
 
   deleteMeetingInvitation owner "example.com" fakeMeetingId removeInvitation >>= assertStatus 404
+
+testMeetingCleanup :: (HasCallStack) => App ()
+testMeetingCleanup = do
+  env <- ask
+  timedOutResult <- liftIO $ timeout (2 * 60 * 1_000_000) $ runAppWithEnv env $ do
+    -- 2 minutes timeout
+    (owner, _tid, _members) <- createTeam OwnDomain 1
+    now <- liftIO getCurrentTime
+    -- Create a meeting that ends now.
+    -- Configured retention is 0.0014 hours (~5 seconds).
+    -- cutoffTime will be now' - 5s.
+    -- We need end_date < cutoffTime.
+    -- If we wait 6 seconds, now' = now + 6s.
+    -- cutoffTime = now + 6s - 5s = now + 1s.
+    -- end_date (now) < cutoffTime (now + 1s).
+    let startTime = addUTCTime (negate 3600) now
+        endTime = now
+        newMeeting = defaultMeetingJson "Cleanup Test" startTime endTime []
+
+    r1 <- postMeetings owner newMeeting
+    assertSuccess r1
+    meeting <- getJSON 201 r1
+    (meetingId, domain) <- getMeetingIdAndDomain meeting
+
+    -- Wait 6 seconds to ensure meeting is old enough
+    liftIO $ threadDelay 6_000_000
+
+    -- Wait for cleanup job to run
+    waitForCleanupJob OwnDomain
+
+    -- Check it's gone
+    getMeeting owner domain meetingId >>= assertStatus 404
+
+  case timedOutResult of
+    Just () -> pure ()
+    Nothing -> assertFailure "testMeetingCleanup timed out after 2 minutes"
+
+waitForCleanupJob :: (HasCallStack, MakesValue domain) => domain -> App ()
+waitForCleanupJob domain = do
+  initialMetrics <- getMetricsBody domain
+  let initialCount = getRunCount initialMetrics
+
+  waitForIncrease domain initialCount
+  where
+    getMetricsBody d = do
+      getMetrics d BackgroundWorker `bindResponse` \resp -> do
+        resp.status `shouldMatchInt` 200
+        pure $ Text.unpack $ Text.decodeUtf8 resp.body
+
+    getRunCount metrics =
+      let (_, _, _, matches) :: (String, String, String, [String]) = (metrics =~ "wire_meetings_cleanup_runs_total ([0-9]+)")
+       in case matches of
+            [val] -> read val :: Int
+            _ -> 0
+
+    waitForIncrease d oldVal = do
+      metrics <- getMetricsBody d
+      let newVal = getRunCount metrics
+      -- We wait until it increases.
+      -- Note: if oldVal was 0 (metric didn't exist), getting 0 again means it hasn't run.
+      -- If it runs, it should become >= 1.
+      -- But wait, if matches is empty, we return 0.
+      -- If the metric appears, it will be >= 1 (initialized at 0? Counter starts at 0).
+      -- If it runs, it increments.
+      when (newVal <= oldVal) $ do
+        liftIO $ threadDelay 1_000_000 -- Wait 1s
+        waitForIncrease d oldVal
+
+testMeetingExpiration :: (HasCallStack) => App ()
+testMeetingExpiration = do
+  (owner, _tid, _members) <- createTeam OwnDomain 1
+  now <- liftIO getCurrentTime
+  let startTime = addUTCTime (negate 3600) now
+      -- meetingValidityPeriodSeconds is configured to 5 seconds in galley.integration.yaml
+      endTime = now
+      newMeeting = defaultMeetingJson "Expiring Meeting" startTime endTime []
+
+  r1 <- postMeetings owner newMeeting
+  assertSuccess r1
+  meeting <- getJSON 201 r1
+  (meetingId, domain) <- getMeetingIdAndDomain meeting
+
+  -- Check it is accessible immediately (endDate = now, so valid until now + 5s)
+  getMeeting owner domain meetingId >>= assertStatus 200
+
+  -- Wait 6 seconds
+  liftIO $ threadDelay 6_000_000
+
+  -- Check it is expired
+  getMeeting owner domain meetingId >>= assertStatus 404

libs/wire-subsystems/src/Wire/MeetingsStore.hs

@@ -164,5 +164,13 @@ data MeetingsStore m a where
     MeetingId ->
     [EmailAddress] ->
     MeetingsStore m ()
+  -- Cleanup operations
+  GetOldMeetings ::
+    UTCTime ->
+    Int ->
+    MeetingsStore m [StoredMeeting]
+  DeleteMeetingBatch ::
+    [MeetingId] ->
+    MeetingsStore m Int64
 
 makeSem ''MeetingsStore

libs/wire-subsystems/src/Wire/MeetingsStore/Postgres.hs

@@ -36,7 +36,7 @@ import Hasql.Statement
 import Hasql.TH
 import Imports
 import Polysemy
-import Polysemy.Error (throw)
+import Polysemy.Error (Error, throw)
 import Polysemy.Input
 import Wire.API.Meeting (Recurrence)
 import Wire.API.PostgresMarshall (PostgresMarshall (..), PostgresUnmarshall (..), dimapPG)
@@ -63,6 +63,10 @@ interpretMeetingsStoreToPostgres =
       addInvitedEmailsImpl meetingId email
     RemoveInvitedEmails meetingId emails ->
       removeInvitedEmailsImpl meetingId emails
+    GetOldMeetings cutoffTime batchSize ->
+      getOldMeetingsImpl cutoffTime batchSize
+    DeleteMeetingBatch meetingIds ->
+      deleteMeetingBatchImpl meetingIds
 
 -- * Create
 
@@ -370,3 +374,56 @@ removeInvitedEmailsImpl meetingId emails = do
             updated_at = NOW()
         WHERE id = ($2 :: uuid)
       |]
+
+getOldMeetingsImpl ::
+  ( Member (Input Pool) r,
+    Member (Embed IO) r,
+    Member (Error UsageError) r
+  ) =>
+  UTCTime ->
+  Int ->
+  Sem r [StoredMeeting]
+getOldMeetingsImpl cutoffTime batchSize = do
+  pool <- input
+  result <- liftIO $ use pool session
+  either throw pure result
+  where
+    session :: Session [StoredMeeting]
+    session = statement (cutoffTime, fromIntegral batchSize) $ V.toList <$> listStatement
+    listStatement :: Statement (UTCTime, Int32) (V.Vector StoredMeeting)
+    listStatement =
+      refineResult
+        (traverse (postgresUnmarshall @StoredMeetingTuple @StoredMeeting))
+        $ [vectorStatement|
+          SELECT
+            id :: uuid, title :: text, creator :: uuid,
+            start_time :: timestamptz, end_time :: timestamptz,
+            recurrence_frequency :: text?, recurrence_interval :: int4?, recurrence_until :: timestamptz?,
+            conversation_id :: uuid, invited_emails :: text[], trial :: boolean,
+            created_at :: timestamptz, updated_at :: timestamptz
+          FROM meetings
+          WHERE end_date < ($1 :: timestamptz)
+          ORDER BY end_date ASC
+          LIMIT ($2 :: int4)
+        |]
+
+deleteMeetingBatchImpl ::
+  ( Member (Input Pool) r,
+    Member (Embed IO) r,
+    Member (Error UsageError) r
+  ) =>
+  [MeetingId] ->
+  Sem r Int64
+deleteMeetingBatchImpl meetingIds = do
+  pool <- input
+  result <- liftIO $ use pool session
+  either throw pure result
+  where
+    session :: Session Int64
+    session = statement (V.fromList (toUUID <$> meetingIds)) deleteStatement
+    deleteStatement :: Statement (V.Vector UUID) Int64
+    deleteStatement =
+      [rowsAffectedStatement|
+        DELETE FROM meetings
+        WHERE (id, domain) IN (SELECT * FROM unnest($1::uuid[]))
+      |]

libs/wire-subsystems/src/Wire/MeetingsSubsystemCleaning.hs

@@ -0,0 +1,32 @@
+{-# LANGUAGE TemplateHaskell #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
+--
+-- 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 <https://www.gnu.org/licenses/>.
+
+module Wire.MeetingsSubsystemCleaning where
+
+import Data.Time.Clock (UTCTime)
+import Imports
+import Polysemy
+
+data MeetingsSubsystemCleaning m a where
+  CleanupOldMeetings ::
+    UTCTime ->
+    Int ->
+    MeetingsSubsystemCleaning m Int64
+
+makeSem ''MeetingsSubsystemCleaning

libs/wire-subsystems/src/Wire/MeetingsSubsystemCleaning/Interpreter.hs

@@ -0,0 +1,73 @@
+{-# LANGUAGE DuplicateRecordFields #-}
+
+-- This file is part of the Wire Server implementation.
+--
+-- Copyright (C) 2026 Wire Swiss GmbH <opensource@wire.com>
+--
+-- 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 <https://www.gnu.org/licenses/>.
+
+module Wire.MeetingsSubsystemCleaning.Interpreter where
+
+import Data.Time.Clock (UTCTime)
+import Imports
+import Polysemy
+import Wire.API.Conversation (GroupConvType (MeetingConversation), cnvmGroupConvType)
+import Wire.ConversationStore qualified as ConvStore
+import Wire.MeetingsStore qualified as Store
+import Wire.MeetingsSubsystemCleaning
+import Wire.StoredConversation (StoredConversation (..))
+
+interpretMeetingsSubsystemCleaning ::
+  ( Member Store.MeetingsStore r,
+    Member ConvStore.ConversationStore r
+  ) =>
+  InterpreterFor MeetingsSubsystemCleaning r
+interpretMeetingsSubsystemCleaning = interpret $ \case
+  CleanupOldMeetings cutoffTime batchSize ->
+    cleanupOldMeetingsImpl cutoffTime batchSize
+
+cleanupOldMeetingsImpl ::
+  ( Member Store.MeetingsStore r,
+    Member ConvStore.ConversationStore r
+  ) =>
+  UTCTime ->
+  Int ->
+  Sem r Int64
+cleanupOldMeetingsImpl cutoffTime batchSize = do
+  -- 1. Fetch old meetings
+  oldMeetings <- Store.getOldMeetings cutoffTime batchSize
+
+  if null oldMeetings
+    then pure 0
+    else do
+      -- 2. Extract meeting IDs and conversation IDs
+      let meetingIds = map (\Store.StoredMeeting {id = mid} -> mid) oldMeetings
+          convIds = map (\Store.StoredMeeting {conversationId = cid} -> cid) oldMeetings
+
+      -- 3. Delete meetings from database
+      deletedCount <- Store.deleteMeetingBatch meetingIds
+
+      -- 4. Delete associated conversations if they are meeting conversations
+      -- We need to check if conversation has GroupConvType = MeetingConversation
+      for_ (zip oldMeetings convIds) $ \(meeting, convId) -> do
+        maybeConv <- ConvStore.getConversation convId
+        case maybeConv of
+          Just conv
+            | conv.metadata.cnvmGroupConvType == Just MeetingConversation,
+              conv.id_ == convId,
+              meeting.conversationId == convId ->
+                ConvStore.deleteConversation convId
+          _ -> pure ()
+
+      pure deletedCount

libs/wire-subsystems/test/unit/Wire/MockInterpreters/MeetingsStore.hs

@@ -102,3 +102,15 @@ inMemoryMeetingsStoreInterpreter = interpret $ \case
                   updatedAt = now
                 }
         modify (Map.insert mid updatedMeeting)
+  GetOldMeetings cutoffTime batchSize ->
+    gets $
+      take batchSize
+        . List.sortOn (.endTime)
+        . filter (\sm -> sm.endTime < cutoffTime)
+        . Map.elems
+  DeleteMeetingBatch meetingIds -> do
+    let deleteOne mid = do
+          exists <- gets (Map.member mid)
+          when exists $ modify (Map.delete mid)
+          pure exists
+    fromIntegral . length <$> traverse deleteOne meetingIds

libs/wire-subsystems/wire-subsystems.cabal

@@ -367,6 +367,8 @@ library
     Wire.MeetingsStore.Postgres
     Wire.MeetingsSubsystem
     Wire.MeetingsSubsystem.Interpreter
+    Wire.MeetingsSubsystemCleaning
+    Wire.MeetingsSubsystemCleaning.Interpreter
     Wire.Migration
     Wire.MigrationLock
     Wire.NotificationSubsystem

services/background-worker/background-worker.cabal

@@ -21,6 +21,7 @@ library
     Wire.BackgroundWorker.Options
     Wire.BackgroundWorker.Util
     Wire.DeadUserNotificationWatcher
+    Wire.MeetingsCleanupWorker
     Wire.PostgresMigrations
 
   hs-source-dirs:     src
@@ -39,6 +40,7 @@ library
     , bytestring-conversion
     , cassandra-util
     , containers
+    , cron
     , data-timeout
     , exceptions
     , extended
@@ -61,6 +63,7 @@ library
     , ssl-util
     , tagged
     , text
+    , time
     , tinylog
     , transformers
     , transformers-base

services/background-worker/background-worker.integration.yaml

@@ -64,6 +64,12 @@ backgroundJobs:
   jobTimeout: 5s
   maxAttempts: 3
 
+# Meetings cleanup configuration for integration
+meetingsCleanup:
+  cleanOlderThanHours: 0.0014  # Clean meetings older than ~5 seconds
+  batchSize: 100
+  schedule: "* * * * *"  # Run every minute
+
 postgresMigration:
   conversation: postgresql
   conversationCodes: postgresql