Add command to remove expired linked devices

service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java

@@ -223,6 +223,7 @@ import org.whispersystems.textsecuregcm.workers.MessagePersisterServiceCommand;
 import org.whispersystems.textsecuregcm.workers.MigrateSignedECPreKeysCommand;
 import org.whispersystems.textsecuregcm.workers.ProcessPushNotificationFeedbackCommand;
 import org.whispersystems.textsecuregcm.workers.RemoveExpiredAccountsCommand;
+import org.whispersystems.textsecuregcm.workers.RemoveExpiredLinkedDevicesCommand;
 import org.whispersystems.textsecuregcm.workers.ScheduledApnPushNotificationSenderServiceCommand;
 import org.whispersystems.textsecuregcm.workers.ServerVersionCommand;
 import org.whispersystems.textsecuregcm.workers.SetRequestLoggingEnabledTask;
@@ -280,6 +281,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
     bootstrap.addCommand(new MigrateSignedECPreKeysCommand());
     bootstrap.addCommand(new RemoveExpiredAccountsCommand(Clock.systemUTC()));
     bootstrap.addCommand(new ProcessPushNotificationFeedbackCommand(Clock.systemUTC()));
+    bootstrap.addCommand(new RemoveExpiredLinkedDevicesCommand());
   }
 
   @Override

service/src/main/java/org/whispersystems/textsecuregcm/storage/Device.java

@@ -8,9 +8,9 @@ package org.whispersystems.textsecuregcm.storage;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import java.time.Duration;
 import java.util.List;
 import java.util.OptionalInt;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import javax.annotation.Nullable;
@@ -28,6 +28,9 @@ public class Device {
   public static final List<Byte> ALL_POSSIBLE_DEVICE_IDS = IntStream.range(Device.PRIMARY_ID, MAXIMUM_DEVICE_ID).boxed()
       .map(Integer::byteValue).collect(Collectors.toList());
 
+  private static final long ALLOWED_LINKED_IDLE_MILLIS = Duration.ofDays(30).toMillis();
+  private static final long ALLOWED_PRIMARY_IDLE_MILLIS = Duration.ofDays(180).toMillis();
+
   @JsonDeserialize(using = DeviceIdDeserializer.class)
   @JsonProperty
   private byte id;
@@ -206,8 +209,13 @@ public class Device {
   public boolean isEnabled() {
     boolean hasChannel = fetchesMessages || StringUtils.isNotEmpty(getApnId()) || StringUtils.isNotEmpty(getGcmId());
 
-    return (id == PRIMARY_ID && hasChannel) ||
-           (id != PRIMARY_ID && hasChannel && lastSeen > (System.currentTimeMillis() - TimeUnit.DAYS.toMillis(30)));
+    return (id == PRIMARY_ID && hasChannel) || (id != PRIMARY_ID && hasChannel && !isExpired());
+  }
+
+  public boolean isExpired() {
+    return isPrimary()
+        ? lastSeen < (System.currentTimeMillis() - ALLOWED_PRIMARY_IDLE_MILLIS)
+        : lastSeen < (System.currentTimeMillis() - ALLOWED_LINKED_IDLE_MILLIS);
   }
 
   public boolean getFetchesMessages() {

service/src/main/java/org/whispersystems/textsecuregcm/workers/RemoveExpiredLinkedDevicesCommand.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2023 Signal Messenger, LLC
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+package org.whispersystems.textsecuregcm.workers;
+
+import static org.whispersystems.textsecuregcm.metrics.MetricsUtil.name;
+
+import io.micrometer.core.instrument.Metrics;
+import io.micrometer.shaded.reactor.util.function.Tuples;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import net.sourceforge.argparse4j.inf.Subparser;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.textsecuregcm.storage.Account;
+import org.whispersystems.textsecuregcm.storage.Device;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+public class RemoveExpiredLinkedDevicesCommand extends AbstractSinglePassCrawlAccountsCommand {
+
+  private static final int MAX_CONCURRENCY = 16;
+
+  private static final String DRY_RUN_ARGUMENT = "dry-run";
+  private static final String REMOVED_DEVICES_COUNTER_NAME = name(RemoveExpiredLinkedDevicesCommand.class,
+      "removedDevices");
+  private static final String UPDATED_ACCOUNTS_COUNTER_NAME = name(RemoveExpiredLinkedDevicesCommand.class,
+      "updatedAccounts");
+  private static final Logger logger = LoggerFactory.getLogger(RemoveExpiredLinkedDevicesCommand.class);
+
+  public RemoveExpiredLinkedDevicesCommand() {
+    super("remove-expired-devices", "Removes expired linked devices");
+  }
+
+  @Override
+  public void configure(final Subparser subparser) {
+    super.configure(subparser);
+
+    subparser.addArgument("--dry-run")
+        .type(Boolean.class)
+        .dest(DRY_RUN_ARGUMENT)
+        .required(false)
+        .setDefault(true)
+        .help("If true, don’t actually modify accounts with expired linked devices");
+  }
+
+  @Override
+  protected void crawlAccounts(final Flux<Account> accounts) {
+
+    final boolean dryRun = getNamespace().getBoolean(DRY_RUN_ARGUMENT);
+
+    accounts.map(a -> Tuples.of(a, getExpiredLinkedDeviceIds(a.getDevices())))
+        .filter(accountAndExpiredDevices -> !accountAndExpiredDevices.getT2().isEmpty())
+        .flatMap(accountAndExpiredDevices -> {
+          final Account account = accountAndExpiredDevices.getT1();
+          final Set<Byte> expiredDevices = accountAndExpiredDevices.getT2();
+
+          final Mono<Void> accountUpdate = dryRun
+              ? Mono.empty()
+              : deleteDevices(account, expiredDevices);
+
+          return accountUpdate.thenReturn(expiredDevices.size())
+              .onErrorResume(t -> {
+                logger.warn("Failed to remove expired linked devices {}", account.getUuid(),
+                    t);
+                return Mono.empty();
+              });
+
+        }, MAX_CONCURRENCY)
+        .doOnNext(removedDevices -> {
+          Metrics.counter(REMOVED_DEVICES_COUNTER_NAME, "dryRun", String.valueOf(dryRun)).increment(removedDevices);
+          Metrics.counter(UPDATED_ACCOUNTS_COUNTER_NAME, "dryRun", String.valueOf(dryRun)).increment();
+        })
+        .then()
+        .block();
+  }
+
+  private Mono<Void> deleteDevices(final Account account, final Set<Byte> expiredDevices) {
+    return Flux.fromIterable(expiredDevices)
+        .flatMap(deviceId ->
+                Mono.fromFuture(() -> getCommandDependencies().accountsManager().removeDevice(account, deviceId)),
+            // limit concurrency to avoid contested updates
+            1)
+        .then();
+  }
+
+  protected static Set<Byte> getExpiredLinkedDeviceIds(List<Device> devices) {
+    return devices.stream()
+        // linked devices
+        .filter(Predicate.not(Device::isPrimary))
+        // that are expired
+        .filter(Device::isExpired)
+        .map(Device::getId)
+        .collect(Collectors.toSet());
+  }
+}