You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

187 lines
6.5 KiB

  1. <?php
  2. use BarnabyWalters\Mf2;
  3. class PushTask {
  4. public static function verify_subscription($subscription_id, $mode) {
  5. $subscription = db\get_by_id('subscriptions', $subscription_id);
  6. if($subscription) {
  7. $feed = db\get_by_id('feeds', $subscription->feed_id);
  8. // Choose the expiration for the subscription
  9. $lease_seconds = 86400*7 + 3600;
  10. $exp_ts = time() + $lease_seconds;
  11. $exp_date = date('Y-m-d H:i:s', $exp_ts);
  12. if($subscription->namespaced) {
  13. $prefix = 'hub.';
  14. } else {
  15. $prefix = '';
  16. }
  17. $push_params = [
  18. $prefix.'mode' => ($mode == 'subscribe' ? 'subscribe' : 'unsubscribe'),
  19. $prefix.'topic' => $feed->feed_url,
  20. $prefix.'challenge' => $subscription->challenge
  21. ];
  22. if($mode == 'subscribe') {
  23. $push_params[$prefix.'lease_seconds'] = $lease_seconds;
  24. }
  25. echo "Verifying subscription to ".$feed->feed_url." for callback ".$subscription->callback_url."\n";
  26. $url = parse_url($subscription->callback_url);
  27. if($q=k($url, 'query')) {
  28. parse_str($q, $existing_params);
  29. $push_params = array_merge($push_params, $existing_params);
  30. }
  31. $url['query'] = http_build_query($push_params);
  32. $url = build_url($url);
  33. echo "\t".$url."\n";
  34. $response = request\get_url($url, true);
  35. $subscription->challenge_response = $response['headers']."\n\n".$response['body'];
  36. if(request\response_is($response['status'], 2) && $response['body'] == $subscription->challenge) {
  37. // The subscriber replied with a 2xx status code and confirmed the challenge string.
  38. if($mode == 'subscribe') {
  39. // The subscription is confirmed and active.
  40. $subscription->date_confirmed = db\now();
  41. $subscription->lease_seconds = $lease_seconds;
  42. $subscription->date_expires = $exp_date;
  43. $subscription->active = 1;
  44. echo "Subscriber verified the request and is now subscribed\n";
  45. } else {
  46. $subscription->date_unsubscribed = db\now();
  47. $subscription->date_expires = null;
  48. $subscription->active = 0;
  49. echo "Subscriber verified the request and is now unsubscribed\n";
  50. }
  51. } else {
  52. // The subscriber did not confirm the subscription, so reject the request
  53. echo "Subscriber did not echo the challenge\n";
  54. }
  55. db\set_updated($subscription);
  56. $subscription->save();
  57. print_r($response);
  58. } else {
  59. echo "Subscription not found\n";
  60. }
  61. }
  62. public static function publish($feed_id) {
  63. $feed = db\get_by_id('feeds', $feed_id);
  64. if($feed) {
  65. // First check the feed to see if the content has changed since the last time we checked
  66. $response = request\get_url($feed->feed_url, true);
  67. $feed->last_retrieved = db\now();
  68. db\set_updated($feed);
  69. $content_hash = md5($response['body']);
  70. if($content_hash != $feed->content_hash) {
  71. $feed->content_hash = $content_hash;
  72. $headers = request\parse_headers($response['headers']);
  73. if(array_key_exists('Content-Type', $headers)) {
  74. $feed->content_type = $headers['Content-Type'];
  75. }
  76. $feed->content = $response['body'];
  77. $feed->save();
  78. $subscribers = ORM::for_table('subscriptions')->where('feed_id', $feed->id)->where('active', 1)->find_many();
  79. foreach($subscribers as $s) {
  80. echo "Queuing notification for feed_id=$feed_id ($feed->feed_url) subscription_id=$s->id ($s->callback_url)\n";
  81. DeferredTask::queue('PushTask', 'notify_subscriber', [$feed_id, $s->id, db\now()]);
  82. }
  83. } else {
  84. $feed->save();
  85. echo "Feed body has the same content hash as last time, not notifying subscribers\n";
  86. }
  87. } else {
  88. echo "Feed not found\n";
  89. }
  90. }
  91. public static function notify_subscriber($feed_id, $subscription_id, $date_queued) {
  92. $feed = db\get_by_id('feeds', $feed_id);
  93. if(!$feed) {
  94. echo "Feed not found\n";
  95. return;
  96. }
  97. $subscription = db\get_by_id('subscriptions', $subscription_id);
  98. if(!$subscription) {
  99. echo "Subscription not found\n";
  100. return;
  101. }
  102. // If the job was put on the queue before the last ping was sent, ignore it.
  103. // This happens when there is a retry job in the delayed queue, and then the
  104. // publisher sends a new publish request and the subscriber responds to it immediately.
  105. if(strtotime($date_queued) < strtotime($subscription->date_last_ping_sent)) {
  106. echo "Job was queued before the last ping was sent by the publisher, skipping\n";
  107. return;
  108. }
  109. echo "Processing subscriber: " . $subscription->callback_url . "\n";
  110. // Subscription may be "active" but the expiration date may have passed.
  111. // If so, de-activate the subscription.
  112. if(strtotime($subscription->date_expires) < time()) {
  113. echo "Subscription expired!\n";
  114. $subscription->active = 0;
  115. db\set_updated($subscription);
  116. $subscription->save();
  117. return;
  118. }
  119. echo "Notifying subscriber!\n";
  120. $subscription->date_last_ping_sent = db\now();
  121. $response = request\post($subscription->callback_url, $feed->content, false, [
  122. 'Content-Type: ' . ($feed->content_type ?: 'text/plain')
  123. ]);
  124. $subscription->last_ping_status = $response['status'];
  125. $subscription->last_ping_headers = $response['headers'];
  126. $subscription->last_ping_body = $response['body'];
  127. echo "Subscriber return a " . $response['status'] . " HTTP status\n";
  128. if(request\response_is($response['status'], 2)) {
  129. $subscription->last_ping_success = 1;
  130. $subscription->last_ping_error_delay = 0;
  131. } else {
  132. $subscription->last_ping_success = 0;
  133. // If the ping failed, queue another ping for a later time with exponential backoff
  134. if($subscription->last_ping_error_delay == 0)
  135. $subscription->last_ping_error_delay = 15;
  136. // If it's timed out after 8 tries, de-activate the subscription
  137. if($subscription->last_ping_error_delay > 2000) {
  138. echo "Ping failed after " . $subscription->last_ping_error_delay . " seconds. Deactivating this subscription.\n";
  139. $subscription->active = 0;
  140. } else {
  141. echo "Ping failed, trying again in " . $subscription->last_ping_error_delay . " seconds\n";
  142. DeferredTask::queue('PushTask', 'notify_subscriber', [$feed_id, $subscription_id, db\now()], $subscription->last_ping_error_delay);
  143. $subscription->last_ping_error_delay = $subscription->last_ping_error_delay * 2;
  144. }
  145. }
  146. db\set_updated($subscription);
  147. $subscription->save();
  148. }
  149. }