Merge pull request #9766 from annando/conversation-direction

Improved direction and protocol detection

mod/dfrn_notify.php

@@ -36,8 +36,6 @@ use Friendica\Util\Network;
 use Friendica\Util\Strings;
 
 function dfrn_notify_post(App $a) {
-	Logger::log(__function__, Logger::TRACE);
-
 	$postdata = Network::postdata();
 
 	if (empty($_POST) || !empty($postdata)) {
@@ -193,7 +191,7 @@ function dfrn_notify_post(App $a) {
 
 	Logger::log('Importing post from ' . $importer['addr'] . ' to ' . $importer['nickname'] . ' with the RINO ' . $rino_remote . ' encryption.', Logger::DEBUG);
 
-	$ret = DFRN::import($data, $importer, false, Conversation::PARCEL_LEGACY_DFRN);
+	$ret = DFRN::import($data, $importer, Conversation::PARCEL_LEGACY_DFRN, Conversation::PUSH);
 	System::xmlExit($ret, 'Processed');
 
 	// NOTREACHED
@@ -225,7 +223,7 @@ function dfrn_dispatch_public($postdata)
 	Logger::log('Importing post from ' . $msg['author'] . ' with the public envelope.', Logger::DEBUG);
 
 	// Now we should be able to import it
-	$ret = DFRN::import($msg['message'], $importer, false, Conversation::PARCEL_DIASPORA_DFRN);
+	$ret = DFRN::import($msg['message'], $importer, Conversation::PARCEL_DIASPORA_DFRN, Conversation::RELAY);
 	System::xmlExit($ret, 'Done');
 }
 
@@ -258,7 +256,7 @@ function dfrn_dispatch_private($user, $postdata)
 	Logger::log('Importing post from ' . $msg['author'] . ' to ' . $user['nickname'] . ' with the private envelope.', Logger::DEBUG);
 
 	// Now we should be able to import it
-	$ret = DFRN::import($msg['message'], $importer, false, Conversation::PARCEL_DIASPORA_DFRN);
+	$ret = DFRN::import($msg['message'], $importer, Conversation::PARCEL_DIASPORA_DFRN, Conversation::PUSH);
 	System::xmlExit($ret, 'Done');
 }
 

mod/events.php

@@ -25,12 +25,13 @@ use Friendica\Content\Nav;
 use Friendica\Content\Widget\CalendarExport;
 use Friendica\Core\ACL;
 use Friendica\Core\Logger;
+use Friendica\Core\Protocol;
 use Friendica\Core\Renderer;
 use Friendica\Core\Theme;
 use Friendica\Core\Worker;
 use Friendica\Database\DBA;
 use Friendica\DI;
-use Friendica\Model\Contact;
+use Friendica\Model\Conversation;
 use Friendica\Model\Event;
 use Friendica\Model\Item;
 use Friendica\Model\User;
@@ -204,6 +205,9 @@ function events_post(App $a)
 	$datarray['deny_gid']  = $str_group_deny;
 	$datarray['private']   = $private_event;
 	$datarray['id']        = $event_id;
+	$datarray['network']   = Protocol::DFRN;
+	$datarray['protocol']  = Conversation::PARCEL_DIRECT;
+	$datarray['direction'] = Conversation::PUSH;
 
 	if (intval($_REQUEST['preview'])) {
 		$html = Event::getHTML($datarray);

mod/item.php

@@ -623,6 +623,7 @@ function item_post(App $a) {
 
 	// This field is for storing the raw conversation data
 	$datarray['protocol'] = Conversation::PARCEL_DIRECT;
+	$datarray['direction'] = Conversation::PUSH;
 
 	$conversation = DBA::selectFirst('conversation', ['conversation-uri', 'conversation-href'], ['item-uri' => $datarray['thr-parent']]);
 	if (DBA::isResult($conversation)) {

mod/pubsub.php

@@ -139,20 +139,15 @@ function pubsub_post(App $a)
 		hub_post_return();
 	}
 
-	// We import feeds from OStatus, Friendica and ATOM/RSS.
+	// We only import feeds from OStatus here
-	/// @todo Check if Friendica posts really arrive here - otherwise we can discard some stuff
+	if ($contact['network'] != Protocol::OSTATUS) {
-	if (!in_array($contact['network'], [Protocol::OSTATUS, Protocol::DFRN, Protocol::FEED])) {
+		Logger::warning('Unexpected network', ['contact' => $contact]);
 		hub_post_return();
 	}
 
 	Logger::log('Import item for ' . $nick . ' from ' . $contact['nick'] . ' (' . $contact['id'] . ')');
 	$feedhub = '';
-	Feed::consume($xml, $importer, $contact, $feedhub);
+	OStatus::import($xml, $importer, $contact, $feedhub);
-
-	// do it a second time for DFRN so that any children find their parents.
-	if ($contact['network'] === Protocol::DFRN) {
-		Feed::consume($xml, $importer, $contact, $feedhub);
-	}
 
 	hub_post_return();
 }

src/Model/Conversation.php

@@ -21,7 +21,6 @@
 
 namespace Friendica\Model;
 
-use Friendica\Core\Logger;
 use Friendica\Core\Protocol;
 use Friendica\Database\Database;
 use Friendica\Database\DBA;
@@ -58,6 +57,10 @@ class Conversation
 	 * The message had been fetched by our system
 	 */
 	const PULL    = 2;
+	/**
+	 * The message had been pushed to this system via a relay server
+	 */
+	const RELAY   = 3;
 
 	public static function getByItemUri($item_uri)
 	{
@@ -105,34 +108,8 @@ class Conversation
 				$conversation['source'] = $arr['source'];
 			}
 
-			$fields = ['item-uri', 'reply-to-uri', 'conversation-uri', 'conversation-href', 'protocol', 'source'];
+			if (!DBA::exists('conversation', ['item-uri' => $conversation['item-uri']])) {
-			$old_conv = DBA::selectFirst('conversation', $fields, ['item-uri' => $conversation['item-uri']]);
+				DBA::insert('conversation', $conversation, Database::INSERT_IGNORE);
-			if (DBA::isResult($old_conv)) {
-				// Don't update when only the source has changed.
-				// Only do this when there had been no source before.
-				if ($old_conv['source'] != '') {
-					unset($old_conv['source']);
-				}
-				// Update structure data all the time but the source only when its from a better protocol.
-				if (
-					empty($conversation['source'])
-					|| (
-						!empty($old_conv['source'])
-						&& ($old_conv['protocol'] < (($conversation['protocol'] ?? '') ?: self::PARCEL_UNKNOWN))
-					)
-				) {
-					unset($conversation['protocol']);
-					unset($conversation['source']);
-				}
-				if (!DBA::update('conversation', $conversation, ['item-uri' => $conversation['item-uri']], $old_conv)) {
-					Logger::log('Conversation: update for ' . $conversation['item-uri'] . ' from ' . $old_conv['protocol'] . ' to ' . $conversation['protocol'] . ' failed',
-						Logger::DEBUG);
-				}
-			} else {
-				if (!DBA::insert('conversation', $conversation, Database::INSERT_UPDATE)) {
-					Logger::log('Conversation: insert for ' . $conversation['item-uri'] . ' (protocol ' . $conversation['protocol'] . ') failed',
-						Logger::DEBUG);
-				}
 			}
 		}
 

src/Model/Event.php

@@ -257,6 +257,16 @@ class Event
 	 */
 	public static function store($arr)
 	{
+		$network = $arr['network'] ?? Protocol::DFRN;
+		$protocol = $arr['protocol'] ?? Conversation::PARCEL_UNKNOWN;
+		$direction = $arr['direction'] ?? Conversation::UNKNOWN;
+		$source = $arr['source'] ?? '';
+
+		unset($arr['network']);
+		unset($arr['protocol']);
+		unset($arr['direction']);
+		unset($arr['source']);
+
 		$event = [];
 		$event['id']        = intval($arr['id']        ?? 0);
 		$event['uid']       = intval($arr['uid']       ?? 0);
@@ -373,7 +383,10 @@ class Event
 				$item_arr['origin']        = $event['cid'] === 0 ? 1 : 0;
 				$item_arr['body']          = self::getBBCode($event);
 				$item_arr['event-id']      = $event['id'];
-				$item_arr['network']       = Protocol::DFRN;
+				$item_arr['network']       = $network;
+				$item_arr['protocol']      = $protocol;
+				$item_arr['direction']     = $direction;
+				$item_arr['source']        = $source;
 
 				$item_arr['object']  = '<object><type>' . XML::escape(Activity\ObjectType::EVENT) . '</type><title></title><id>' . XML::escape($event['uri']) . '</id>';
 				$item_arr['object'] .= '<content>' . XML::escape(self::getBBCode($event)) . '</content>';

src/Model/Item.php

@@ -1581,6 +1581,7 @@ class Item
 			$item['origin'] = 1;
 			$item['network'] = Protocol::DFRN;
 			$item['protocol'] = Conversation::PARCEL_DIRECT;
+			$item['direction'] = Conversation::PUSH;
 
 			if (in_array($notify, PRIORITIES)) {
 				$priority = $notify;
@@ -3336,6 +3337,8 @@ class Item
 			'wall'          => $item['wall'],
 			'origin'        => 1,
 			'network'       => Protocol::DFRN,
+			'protocol'      => Conversation::PARCEL_DIRECT,
+			'direction'     => Conversation::PUSH,
 			'gravity'       => GRAVITY_ACTIVITY,
 			'parent'        => $item['id'],
 			'thr-parent'    => $item['uri'],

src/Protocol/ActivityPub/Processor.php

@@ -297,6 +297,10 @@ class Processor
 			}
 		}
 
+		if (!empty($activity['from-relay'])) {
+			$item['direction'] = Conversation::RELAY;
+		}
+
 		$item['isForum'] = false;
 
 		if (!empty($activity['thread-completion'])) {
@@ -420,20 +424,24 @@ class Processor
 	 */
 	public static function createEvent($activity, $item)
 	{
-		$event['summary']  = HTML::toBBCode($activity['name']);
+		$event['summary']   = HTML::toBBCode($activity['name']);
-		$event['desc']     = HTML::toBBCode($activity['content']);
+		$event['desc']      = HTML::toBBCode($activity['content']);
-		$event['start']    = $activity['start-time'];
+		$event['start']     = $activity['start-time'];
-		$event['finish']   = $activity['end-time'];
+		$event['finish']    = $activity['end-time'];
-		$event['nofinish'] = empty($event['finish']);
+		$event['nofinish']  = empty($event['finish']);
-		$event['location'] = $activity['location'];
+		$event['location']  = $activity['location'];
-		$event['adjust']   = true;
+		$event['adjust']    = true;
-		$event['cid']      = $item['contact-id'];
+		$event['cid']       = $item['contact-id'];
-		$event['uid']      = $item['uid'];
+		$event['uid']       = $item['uid'];
-		$event['uri']      = $item['uri'];
+		$event['uri']       = $item['uri'];
-		$event['edited']   = $item['edited'];
+		$event['edited']    = $item['edited'];
-		$event['private']  = $item['private'];
+		$event['private']   = $item['private'];
-		$event['guid']     = $item['guid'];
+		$event['guid']      = $item['guid'];
-		$event['plink']    = $item['plink'];
+		$event['plink']     = $item['plink'];
+		$event['network']   = $item['network'];
+		$event['protocol']  = $item['protocol'];
+		$event['direction'] = $item['direction'];
+		$event['source']    = $item['source'];
 
 		$condition = ['uri' => $item['uri'], 'uid' => $item['uid']];
 		$ev = DBA::selectFirst('event', ['id'], $condition);

src/Protocol/ActivityPub/Receiver.php

@@ -358,6 +358,7 @@ class Receiver
 			$object_data['author'] = JsonLD::fetchElement($activity, 'as:actor', '@id');
 			$object_data['object_id'] = $object_id;
 			$object_data['object_type'] = ''; // Since we don't fetch the object, we don't know the type
+			$object_data['push'] = $push;
 		} elseif (in_array($type, ['as:Add'])) {
 			$object_data = [];
 			$object_data['id'] = JsonLD::fetchElement($activity, '@id');
@@ -365,6 +366,7 @@ class Receiver
 			$object_data['object_id'] = JsonLD::fetchElement($activity, 'as:object', '@id');
 			$object_data['object_type'] = JsonLD::fetchElement($activity['as:object'], '@type');
 			$object_data['object_content'] = JsonLD::fetchElement($activity['as:object'], 'as:content', '@type');
+			$object_data['push'] = $push;
 		} else {
 			$object_data = [];
 			$object_data['id'] = JsonLD::fetchElement($activity, '@id');
@@ -372,6 +374,7 @@ class Receiver
 			$object_data['object_actor'] = JsonLD::fetchElement($activity['as:object'], 'as:actor', '@id');
 			$object_data['object_object'] = JsonLD::fetchElement($activity['as:object'], 'as:object');
 			$object_data['object_type'] = JsonLD::fetchElement($activity['as:object'], '@type');
+			$object_data['push'] = $push;
 
 			// An Undo is done on the object of an object, so we need that type as well
 			if (($type == 'as:Undo') && !empty($object_data['object_object'])) {

src/Protocol/DFRN.php

@@ -2425,13 +2425,17 @@ class DFRN
 				$ev = Event::fromBBCode($item["body"]);
 				if ((!empty($ev['desc']) || !empty($ev['summary'])) && !empty($ev['start'])) {
 					Logger::log("Event in item ".$item["uri"]." was found.", Logger::DEBUG);
-					$ev["cid"]     = $importer["id"];
+					$ev["cid"]       = $importer["id"];
-					$ev["uid"]     = $importer["importer_uid"];
+					$ev["uid"]       = $importer["importer_uid"];
-					$ev["uri"]     = $item["uri"];
+					$ev["uri"]       = $item["uri"];
-					$ev["edited"]  = $item["edited"];
+					$ev["edited"]    = $item["edited"];
-					$ev["private"] = $item["private"];
+					$ev["private"]   = $item["private"];
-					$ev["guid"]    = $item["guid"];
+					$ev["guid"]      = $item["guid"];
-					$ev["plink"]   = $item["plink"];
+					$ev["plink"]     = $item["plink"];
+					$ev["network"]   = $item["network"];
+					$ev["protocol"]  = $item["protocol"];
+					$ev["direction"] = $item["direction"];
+					$ev["source"]    = $item["source"];
 
 					$condition = ['uri' => $item["uri"], 'uid' => $importer["importer_uid"]];
 					$event = DBA::selectFirst('event', ['id'], $condition);
@@ -2593,15 +2597,16 @@ class DFRN
 	/**
 	 * Imports a DFRN message
 	 *
-	 * @param string $xml          The DFRN message
+	 * @param string $xml       The DFRN message
-	 * @param array  $importer     Record of the importer user mixed with contact of the content
+	 * @param array  $importer  Record of the importer user mixed with contact of the content
-	 * @param bool   $sort_by_date Is used when feeds are polled
+	 * @param int    $protocol  Transport protocol
+	 * @param int    $direction Is the message pushed or pulled?
 	 * @return integer Import status
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @throws \ImagickException
 	 * @todo  set proper type-hints
 	 */
-	public static function import($xml, $importer, $sort_by_date = false, $protocol = Conversation::PARCEL_DFRN)
+	public static function import($xml, $importer, $protocol, $direction)
 	{
 		if ($xml == "") {
 			return 400;
@@ -2628,6 +2633,11 @@ class DFRN
 		$header["wall"] = 0;
 		$header["origin"] = 0;
 		$header["contact-id"] = $importer["id"];
+		$header["direction"] = $direction;
+
+		if ($direction === Conversation::RELAY) {
+			$header['post-type'] = Item::PT_RELAY;
+		}
 
 		// Update the contact table if the data has changed
 
@@ -2715,26 +2725,11 @@ class DFRN
 			}
 		}
 
-		if (!$sort_by_date) {
+		$entries = $xpath->query("/atom:feed/atom:entry");
-			$entries = $xpath->query("/atom:feed/atom:entry");
+		foreach ($entries as $entry) {
-			foreach ($entries as $entry) {
+			self::processEntry($header, $xpath, $entry, $importer, $xml, $protocol);
-				self::processEntry($header, $xpath, $entry, $importer, $xml, $protocol);
-			}
-		} else {
-			$newentries = [];
-			$entries = $xpath->query("/atom:feed/atom:entry");
-			foreach ($entries as $entry) {
-				$created = XML::getFirstNodeValue($xpath, "atom:published/text()", $entry);
-				$newentries[strtotime($created)] = $entry;
-			}
-
-			// Now sort after the publishing date
-			ksort($newentries);
-
-			foreach ($newentries as $entry) {
-				self::processEntry($header, $xpath, $entry, $importer, $xml, $protocol);
-			}
 		}
+
 		Logger::log("Import done for user " . $importer["importer_uid"] . " from contact " . $importer["id"], Logger::DEBUG);
 		return 200;
 	}

src/Protocol/Diaspora.php

@@ -537,7 +537,7 @@ class Diaspora
 				return self::receiveConversation($importer, $msg, $fields);
 
 			case "like":
-				return self::receiveLike($importer, $sender, $fields);
+				return self::receiveLike($importer, $sender, $fields, $fetched);
 
 			case "message":
 				if (!$private) {
@@ -551,7 +551,7 @@ class Diaspora
 					Logger::log('Message with type ' . $type . ' is not private, quitting.');
 					return false;
 				}
-				return self::receiveParticipation($importer, $fields);
+				return self::receiveParticipation($importer, $fields, $fetched);
 
 			case "photo": // Not implemented
 				return self::receivePhoto($importer, $fields);
@@ -567,7 +567,7 @@ class Diaspora
 				return self::receiveProfile($importer, $fields);
 
 			case "reshare":
-				return self::receiveReshare($importer, $fields, $msg["message"]);
+				return self::receiveReshare($importer, $fields, $msg["message"], $fetched);
 
 			case "retraction":
 				return self::receiveRetraction($importer, $sender, $fields);
@@ -1570,6 +1570,7 @@ class Diaspora
 
 		$datarray["protocol"] = Conversation::PARCEL_DIASPORA;
 		$datarray["source"] = $xml;
+		$datarray["direction"] = $fetched ? Conversation::PULL : Conversation::PUSH;
 
 		$datarray["changed"] = $datarray["created"] = $datarray["edited"] = $created_at;
 
@@ -1740,7 +1741,7 @@ class Diaspora
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @throws \ImagickException
 	 */
-	private static function receiveLike(array $importer, $sender, $data)
+	private static function receiveLike(array $importer, $sender, $data, bool $fetched)
 	{
 		$author = Strings::escapeTags(XML::unescape($data->author));
 		$guid = Strings::escapeTags(XML::unescape($data->guid));
@@ -1789,6 +1790,7 @@ class Diaspora
 		$datarray = [];
 
 		$datarray["protocol"] = Conversation::PARCEL_DIASPORA;
+		$datarray["direction"] = $fetched ? Conversation::PULL : Conversation::PUSH;
 
 		$datarray["uid"] = $importer["uid"];
 		$datarray["contact-id"] = $author_contact["cid"];
@@ -1917,7 +1919,7 @@ class Diaspora
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @throws \ImagickException
 	 */
-	private static function receiveParticipation(array $importer, $data)
+	private static function receiveParticipation(array $importer, $data, bool $fetched)
 	{
 		$author = strtolower(Strings::escapeTags(XML::unescape($data->author)));
 		$guid = Strings::escapeTags(XML::unescape($data->guid));
@@ -1958,6 +1960,7 @@ class Diaspora
 		$datarray = [];
 
 		$datarray["protocol"] = Conversation::PARCEL_DIASPORA;
+		$datarray["direction"] = $fetched ? Conversation::PULL : Conversation::PUSH;
 
 		$datarray["uid"] = $importer["uid"];
 		$datarray["contact-id"] = $author_contact["cid"];
@@ -2375,6 +2378,8 @@ class Diaspora
 		$datarray['object-type'] = Activity\ObjectType::NOTE;
 
 		$datarray['protocol'] = $item['protocol'];
+		$datarray['source'] = $item['source'];
+		$datarray['direction'] = $item['direction'];
 
 		$datarray['plink'] = self::plink($author, $datarray['guid']);
 		$datarray['private'] = $item['private'];
@@ -2406,7 +2411,7 @@ class Diaspora
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @throws \ImagickException
 	 */
-	private static function receiveReshare(array $importer, $data, $xml)
+	private static function receiveReshare(array $importer, $data, $xml, bool $fetched)
 	{
 		$author = Strings::escapeTags(XML::unescape($data->author));
 		$guid = Strings::escapeTags(XML::unescape($data->guid));
@@ -2452,6 +2457,7 @@ class Diaspora
 
 		$datarray["protocol"] = Conversation::PARCEL_DIASPORA;
 		$datarray["source"] = $xml;
+		$datarray["direction"] = $fetched ? Conversation::PULL : Conversation::PUSH;
 
 		/// @todo Copy tag data from original post
 
@@ -2749,6 +2755,7 @@ class Diaspora
 
 		$datarray["protocol"] = Conversation::PARCEL_DIASPORA;
 		$datarray["source"] = $xml;
+		$datarray["direction"] = $fetched ? Conversation::PULL : Conversation::PUSH;
 
 		if ($fetched) {
 			$datarray["post-type"] = Item::PT_FETCHED;

src/Protocol/Feed.php

@@ -49,65 +49,6 @@ use Friendica\Util\XML;
  */
 class Feed
 {
-	/**
-	 * consume - process atom feed and update anything/everything we might need to update
-	 *
-	 * $xml = the (atom) feed to consume - RSS isn't as fully supported but may work for simple feeds.
-	 *
-	 * $importer = the contact_record (joined to user_record) of the local user who owns this relationship.
-	 *             It is this person's stuff that is going to be updated.
-	 * $contact =  the person who is sending us stuff. If not set, we MAY be processing a "follow" activity
-	 *             from an external network and MAY create an appropriate contact record. Otherwise, we MUST
-	 *             have a contact record.
-	 * $hub = should we find a hub declation in the feed, pass it back to our calling process, who might (or
-	 *        might not) try and subscribe to it.
-	 * $datedir sorts in reverse order
-	 * $pass - by default ($pass = 0) we cannot guarantee that a parent item has been
-	 *      imported prior to its children being seen in the stream unless we are certain
-	 *      of how the feed is arranged/ordered.
-	 * With $pass = 1, we only pull parent items out of the stream.
-	 * With $pass = 2, we only pull children (comments/likes).
-	 *
-	 * So running this twice, first with pass 1 and then with pass 2 will do the right
-	 * thing regardless of feed ordering. This won't be adequate in a fully-threaded
-	 * model where comments can have sub-threads. That would require some massive sorting
-	 * to get all the feed items into a mostly linear ordering, and might still require
-	 * recursion.
-	 *
-	 * @param       $xml
-	 * @param array $importer
-	 * @param array $contact
-	 * @param       $hub
-	 * @throws ImagickException
-	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
-	 */
-	public static function consume($xml, array $importer, array $contact, &$hub)
-	{
-		if ($contact['network'] === Protocol::OSTATUS) {
-			Logger::info('Consume OStatus messages');
-			OStatus::import($xml, $importer, $contact, $hub);
-
-			return;
-		}
-
-		if ($contact['network'] === Protocol::FEED) {
-			Logger::info('Consume feeds');
-			self::import($xml, $importer, $contact);
-
-			return;
-		}
-
-		if ($contact['network'] === Protocol::DFRN) {
-			Logger::info('Consume DFRN messages');
-			$dfrn_importer = DFRN::getImporter($contact['id'], $importer['uid']);
-			if (!empty($dfrn_importer)) {
-				Logger::info('Now import the DFRN feed');
-				DFRN::import($xml, $dfrn_importer, true, Conversation::PARCEL_LEGACY_DFRN);
-				return;
-			}
-		}
-	}
-
 	/**
 	 * Read a RSS/RDF/Atom feed and create an item entry for it
 	 *

src/Protocol/OStatus.php

@@ -312,7 +312,7 @@ class OStatus
 	 */
 	public static function import($xml, array $importer, array &$contact, &$hub)
 	{
-		self::process($xml, $importer, $contact, $hub);
+		self::process($xml, $importer, $contact, $hub, false, true, Conversation::PUSH);
 	}
 
 	/**
@@ -329,7 +329,7 @@ class OStatus
 	 * @throws \Friendica\Network\HTTPException\InternalServerErrorException
 	 * @throws \ImagickException
 	 */
-	private static function process($xml, array $importer, array &$contact = null, &$hub, $stored = false, $initialize = true)
+	private static function process($xml, array $importer, array &$contact = null, &$hub, $stored = false, $initialize = true, $direction = Conversation::UNKNOWN)
 	{
 		if ($initialize) {
 			self::$itemlist = [];
@@ -397,6 +397,7 @@ class OStatus
 
 			$header["protocol"] = Conversation::PARCEL_SALMON;
 			$header["source"] = $xml2;
+			$header["direction"] = $direction;
 		} elseif (!$initialize) {
 			return false;
 		}
@@ -807,6 +808,7 @@ class OStatus
 			$conv_data = [];
 
 			$conv_data['protocol'] = Conversation::PARCEL_SPLIT_CONVERSATION;
+			$conv_data['direction'] = Conversation::PULL;
 			$conv_data['network'] = Protocol::OSTATUS;
 			$conv_data['uri'] = XML::getFirstNodeValue($xpath, 'atom:id/text()', $entry);
 
@@ -847,12 +849,6 @@ class OStatus
 
 			$conv_data['source'] = $doc2->saveXML();
 
-			$condition = ['item-uri' => $conv_data['uri'],'protocol' => Conversation::PARCEL_FEED];
-			if (DBA::exists('conversation', $condition)) {
-				Logger::log('Delete deprecated entry for URI '.$conv_data['uri'], Logger::DEBUG);
-				DBA::delete('conversation', ['item-uri' => $conv_data['uri']]);
-			}
-
 			Logger::log('Store conversation data for uri '.$conv_data['uri'], Logger::DEBUG);
 			Conversation::insert($conv_data);
 		}
@@ -895,6 +891,7 @@ class OStatus
 
 		$item["protocol"] = Conversation::PARCEL_SALMON;
 		$item["source"] = $xml;
+		$item["direction"] = Conversation::PULL;
 
 		Logger::log('Conversation '.$item['uri'].' is now fetched.', Logger::DEBUG);
 	}
@@ -918,7 +915,7 @@ class OStatus
 		if (DBA::isResult($conversation)) {
 			$stored = true;
 			$xml = $conversation['source'];
-			if (self::process($xml, $importer, $contact, $hub, $stored, false)) {
+			if (self::process($xml, $importer, $contact, $hub, $stored, false, Conversation::PULL)) {
 				Logger::log('Got valid cached XML for URI '.$related_uri, Logger::DEBUG);
 				return;
 			}
@@ -1003,7 +1000,7 @@ class OStatus
 		}
 
 		if ($xml != '') {
-			self::process($xml, $importer, $contact, $hub, $stored, false);
+			self::process($xml, $importer, $contact, $hub, $stored, false, Conversation::PULL);
 		} else {
 			Logger::log("XML couldn't be fetched for URI: ".$related_uri." - href: ".$related, Logger::DEBUG);
 		}

src/Worker/Delivery.php

@@ -334,7 +334,7 @@ class Delivery
 				return;
 			}
 
-			DFRN::import($atom, $target_importer, false, Conversation::PARCEL_LOCAL_DFRN);
+			DFRN::import($atom, $target_importer, Conversation::PARCEL_LOCAL_DFRN, Conversation::PUSH);
 
 			if (in_array($cmd, [Delivery::POST, Delivery::POKE])) {
 				Model\Post\DeliveryData::incrementQueueDone($target_item['uri-id'], Model\Post\DeliveryData::DFRN);