Parallelize Notifications

lib/Listener/CalendarListener.php

@@ -37,6 +37,7 @@
 use OCA\DAV\Events\CardUpdatedEvent;
 
 use Psr\Log\LoggerInterface;
+use GuzzleHttp\Promise;
 
 use OCA\DavPush\Service\SubscriptionService;
 use OCA\DavPush\Transport\TransportManager;
@@ -59,13 +60,26 @@ public function handle(Event $event): void {
 		$collectionName = $event->getCalendarData()['uri'];
 		$subscriptions = $this->subscriptionService->findAll($collectionName);
 
-		foreach($subscriptions as $subscription) {
-			$transport = $this->transportManager->getTransport($subscription->getTransport());
-			try {
-				$transport->notify($subscription->getUserId(), $collectionName, $subscription->getId());
-			} catch (\Exception $e) {
-				$this->logger->error("transport " .  $subscription->getTransport() . " failed to deliver notification to subscription " . $subscription->getId());
+		$logger = $this->logger;
+
+		$notificationPromises = (function () use ($collectionName, $subscriptions, $logger): \Generator {
+			foreach($subscriptions as $subscription) {
+				$transport = $this->transportManager->getTransport($subscription->getTransport());
+
+				yield $transport->notify($subscription->getUserId(), $collectionName, $subscription->getId())
+					->then(function($result) use ($subscription, $logger) {
+						$logger->debug("transport " . $subscription->getTransport() . " successfully delivered notification for subscription " . $subscription->getId());
+					}, function ($reason) use ($subscription, $logger) {
+						$logger->error("transport " . $subscription->getTransport() . " failed to deliver notification to subscription " . $subscription->getId() . " with error message: " . $reason);
+						return true;
+					});
 			}
+		})();
+
+		try {
+			$responses = Promise\Utils::settle($notificationPromises)->wait(false);
+		} catch (\Exception $e) {
+			$logger->error("HOW DID WE GET HERE; THIS IS NOT HOW THIS IS SUPPOSED TO WORK " . $e->getMessage());
 		}
     }
 }

lib/Migration/Version001Date20240515221000.php

@@ -67,6 +67,9 @@ public function changeSchema(IOutput $output, Closure $schemaClosure, array $opt
 			$table->addColumn('creation_timestamp', Types::BIGINT, [
 				'notnull' => true,
 			]);
+			$table->addColumn('last_update_timestamp', Types::BIGINT, [
+				'notnull' => true,
+			]);
 			$table->addColumn('expiration_timestamp', Types::BIGINT, [
 				'notnull' => true,
 			]);

lib/PushTransports/WebPushTransport.php

@@ -30,13 +30,17 @@
 use OCA\DavPush\Service\WebPushSubscriptionService;
 use OCA\DavPush\Errors\WebPushSubscriptionNotFound;
 
+use OCP\Http\Client\IClientService;
+use OCP\Http\Client\IPromise;
+
 use Sabre\Xml\Service;
 
 class WebPushTransport extends Transport {
 	protected $id = "web-push";
 
 	public function __construct(
 		private WebPushSubscriptionService $webPushSubscriptionService,
+		private IClientService $httpClientService
 	) {}
 
 	private function parseOptions(array $options): array {
@@ -79,7 +83,7 @@ public function registerSubscription($subsciptionId, $options) {
 		];
 	}
 
-	public function notify(string $userId, string $collectionName, int $subscriptionId) {
+	public function notify(string $userId, string $collectionName, int $subscriptionId): IPromise {
 		$xmlService = new Service();
 
 		$pushResource = $this->webPushSubscriptionService->findBySubscriptionId($subscriptionId)->getPushResource();
@@ -88,15 +92,15 @@ public function notify(string $userId, string $collectionName, int $subscription
 			'{DAV:Push}topic' => $collectionName,
 		]);
 
-		$options = [
-			'http' => [
-				'method' => 'POST',
-				'content' => $content,
-			],
-		];
+		$httpClient = $this->httpClientService->newClient();
 
-		$context = stream_context_create($options);
-		$result = file_get_contents($pushResource, false, $context);
+		return $httpClient->postAsync($pushResource, [
+			"body" => $content,
+			"timeout" => 10,
+			"headers" => [
+				"Content-Type" => "application/xml",
+			],
+		]);
 	}
 
 	public function getSubscriptionIdFromOptions(string $userId, string $collectionName, $options): ?int {

lib/Transport/Transport.php

@@ -26,6 +26,8 @@
 
 namespace OCA\DavPush\Transport;
 
+use OCP\Http\Client\IPromise;
+
 abstract class Transport {
 	protected $id;
 
@@ -69,5 +71,5 @@ abstract public function getSubscriptionIdFromOptions(string $userId, string $co
 	// Change mutable options of the subscription (if any exist)
 	abstract public function updateSubscription($subsciptionId, $options);
 
-	abstract public function notify(string $userId, string $collectionName, int $subscriptionId);
+	abstract public function notify(string $userId, string $collectionName, int $subscriptionId): IPromise;
 }