mirror of
https://github.com/friendica/friendica
synced 2025-05-12 07:04:09 +02:00
Merge branch '2024.09-rc' into merge-2024.09-rc-into-develop
This commit is contained in:
Art4
commit
c6c29b29b1
20 changed files with 554 additions and 146 deletions
|
|
@ -66,6 +66,11 @@ final class ATProtocol
|
|||
$this->httpClient = $httpClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns an array of user ids who want to import the Bluesky timeline
|
||||
*
|
||||
* @return array user ids
|
||||
*/
|
||||
public function getUids(): array
|
||||
{
|
||||
$uids = [];
|
||||
|
|
@ -92,6 +97,15 @@ final class ATProtocol
|
|||
return $uids;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches XRPC data
|
||||
* @see https://atproto.com/specs/xrpc#lexicon-http-endpoints
|
||||
*
|
||||
* @param string $url for example "app.bsky.feed.getTimeline"
|
||||
* @param array $parameters Array with parameters
|
||||
* @param integer $uid User ID
|
||||
* @return stdClass|null Fetched data
|
||||
*/
|
||||
public function XRPCGet(string $url, array $parameters = [], int $uid = 0): ?stdClass
|
||||
{
|
||||
if (!empty($parameters)) {
|
||||
|
|
@ -119,6 +133,13 @@ final class ATProtocol
|
|||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch data from the given URL via GET and return it as a JSON class
|
||||
*
|
||||
* @param string $url HTTP URL
|
||||
* @param array $opts HTTP options
|
||||
* @return stdClass|null Fetched data
|
||||
*/
|
||||
public function get(string $url, array $opts = []): ?stdClass
|
||||
{
|
||||
try {
|
||||
|
|
@ -141,13 +162,31 @@ final class ATProtocol
|
|||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Perform an XRPC post for a given user
|
||||
* @see https://atproto.com/specs/xrpc#lexicon-http-endpoints
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @param string $url Endpoints like "com.atproto.repo.createRecord"
|
||||
* @param [type] $parameters array or StdClass with parameters
|
||||
* @return stdClass|null
|
||||
*/
|
||||
public function XRPCPost(int $uid, string $url, $parameters): ?stdClass
|
||||
{
|
||||
$data = $this->post($uid, '/xrpc/' . $url, json_encode($parameters), ['Content-type' => 'application/json', 'Authorization' => ['Bearer ' . $this->getUserToken($uid)]]);
|
||||
return $data;
|
||||
}
|
||||
|
||||
private function post(int $uid, string $url, string $params, array $headers): ?stdClass
|
||||
/**
|
||||
* Post data to the user PDS
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @param string $url HTTP URL without the hostname
|
||||
* @param string $params Parameter string
|
||||
* @param array $headers HTTP header information
|
||||
* @return stdClass|null
|
||||
*/
|
||||
public function post(int $uid, string $url, string $params, array $headers): ?stdClass
|
||||
{
|
||||
$pds = $this->getUserPds($uid);
|
||||
if (empty($pds)) {
|
||||
|
|
@ -172,11 +211,21 @@ final class ATProtocol
|
|||
$data->code = $curlResult->getReturnCode();
|
||||
}
|
||||
|
||||
$this->pConfig->set($uid, 'bluesky', 'status', self::STATUS_SUCCESS);
|
||||
Item::incrementOutbound(Protocol::BLUESKY);
|
||||
if (!empty($data->code) && ($data->code >= 200) && ($data->code < 400)) {
|
||||
$this->pConfig->set($uid, 'bluesky', 'status', self::STATUS_SUCCESS);
|
||||
} else {
|
||||
$this->pConfig->set($uid, 'bluesky', 'status', self::STATUS_API_FAIL);
|
||||
}
|
||||
return $data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches the PDS for a given user
|
||||
* @see https://atproto.com/guides/glossary#pds-personal-data-server
|
||||
*
|
||||
* @param integer $uid User ID or 0
|
||||
* @return string|null PDS or null if the user has got no PDS assigned. If UID set to 0, the public api URL is used
|
||||
*/
|
||||
private function getUserPds(int $uid): ?string
|
||||
{
|
||||
if ($uid == 0) {
|
||||
|
|
@ -202,6 +251,14 @@ final class ATProtocol
|
|||
return $pds;
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch the DID for a given user
|
||||
* @see https://atproto.com/guides/glossary#did-decentralized-id
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @param boolean $refresh Default "false". If set to true, the DID is detected from the handle again.
|
||||
* @return string|null DID or null if no DID has been found.
|
||||
*/
|
||||
public function getUserDid(int $uid, bool $refresh = false): ?string
|
||||
{
|
||||
if (!$this->pConfig->get($uid, 'bluesky', 'post')) {
|
||||
|
|
@ -230,7 +287,13 @@ final class ATProtocol
|
|||
return $did;
|
||||
}
|
||||
|
||||
private function getDid(string $handle): string
|
||||
/**
|
||||
* Fetches the DID for a given handle
|
||||
*
|
||||
* @param string $handle The user handle
|
||||
* @return string DID (did:plc:...)
|
||||
*/
|
||||
public function getDid(string $handle): string
|
||||
{
|
||||
if ($handle == '') {
|
||||
return '';
|
||||
|
|
@ -265,6 +328,12 @@ final class ATProtocol
|
|||
return '';
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches a DID for a given profile URL
|
||||
*
|
||||
* @param string $url HTTP path to the profile in the format https://bsky.app/profile/username
|
||||
* @return string DID (did:plc:...)
|
||||
*/
|
||||
public function getDidByProfile(string $url): string
|
||||
{
|
||||
if (preg_match('#^' . self::WEB . '/profile/(.+)#', $url, $matches)) {
|
||||
|
|
@ -314,6 +383,13 @@ final class ATProtocol
|
|||
return $ids['bsky_did'];
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches the DID of a given handle via a HTTP request to the .well-known URL.
|
||||
* This is one of the ways, custom handles can be authorized.
|
||||
*
|
||||
* @param string $handle The user handle
|
||||
* @return string DID (did:plc:...)
|
||||
*/
|
||||
private function getDidByWellknown(string $handle): string
|
||||
{
|
||||
$curlResult = $this->httpClient->get('http://' . $handle . '/.well-known/atproto-did');
|
||||
|
|
@ -328,6 +404,13 @@ final class ATProtocol
|
|||
return '';
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetches the DID of a given handle via a DND request.
|
||||
* This is one of the ways, custom handles can be authorized.
|
||||
*
|
||||
* @param string $handle The user handle
|
||||
* @return string DID (did:plc:...)
|
||||
*/
|
||||
private function getDidByDns(string $handle): string
|
||||
{
|
||||
$records = @dns_get_record('_atproto.' . $handle . '.', DNS_TXT);
|
||||
|
|
@ -347,7 +430,13 @@ final class ATProtocol
|
|||
return '';
|
||||
}
|
||||
|
||||
private function getPdsOfDid(string $did): ?string
|
||||
/**
|
||||
* Fetch the PDS of a given DID
|
||||
*
|
||||
* @param string $did DID (did:plc:...)
|
||||
* @return string|null URL of the PDS, e.g. https://enoki.us-east.host.bsky.network
|
||||
*/
|
||||
public function getPdsOfDid(string $did): ?string
|
||||
{
|
||||
$data = $this->get(self::DIRECTORY . '/' . $did);
|
||||
if (empty($data) || empty($data->service)) {
|
||||
|
|
@ -363,6 +452,13 @@ final class ATProtocol
|
|||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks if the provided DID matches the handle
|
||||
*
|
||||
* @param string $did DID (did:plc:...)
|
||||
* @param string $handle The user handle
|
||||
* @return boolean
|
||||
*/
|
||||
private function isValidDid(string $did, string $handle): bool
|
||||
{
|
||||
$data = $this->get(self::DIRECTORY . '/' . $did);
|
||||
|
|
@ -373,7 +469,13 @@ final class ATProtocol
|
|||
return in_array('at://' . $handle, $data->alsoKnownAs);
|
||||
}
|
||||
|
||||
private function getUserToken(int $uid): string
|
||||
/**
|
||||
* Fetches the user token for a given user
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @return string user token
|
||||
*/
|
||||
public function getUserToken(int $uid): string
|
||||
{
|
||||
$token = $this->pConfig->get($uid, 'bluesky', 'access_token');
|
||||
$created = $this->pConfig->get($uid, 'bluesky', 'token_created');
|
||||
|
|
@ -387,12 +489,23 @@ final class ATProtocol
|
|||
return $token;
|
||||
}
|
||||
|
||||
/**
|
||||
* Refresh and returns the user token for a given user.
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @return string user token
|
||||
*/
|
||||
private function refreshUserToken(int $uid): string
|
||||
{
|
||||
$token = $this->pConfig->get($uid, 'bluesky', 'refresh_token');
|
||||
|
||||
$data = $this->post($uid, '/xrpc/com.atproto.server.refreshSession', '', ['Authorization' => ['Bearer ' . $token]]);
|
||||
if (empty($data) || empty($data->accessJwt)) {
|
||||
$this->logger->debug('Refresh failed', ['return' => $data]);
|
||||
$password = $this->pConfig->get($uid, 'bluesky', 'password');
|
||||
if (!empty($password)) {
|
||||
return $this->createUserToken($uid, $password);
|
||||
}
|
||||
$this->pConfig->set($uid, 'bluesky', 'status', self::STATUS_TOKEN_FAIL);
|
||||
return '';
|
||||
}
|
||||
|
|
@ -404,6 +517,13 @@ final class ATProtocol
|
|||
return $data->accessJwt;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a user token for the given user
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @param string $password Application password
|
||||
* @return string user token
|
||||
*/
|
||||
public function createUserToken(int $uid, string $password): string
|
||||
{
|
||||
$did = $this->getUserDid($uid);
|
||||
|
|
|
|||
|
|
@ -35,7 +35,13 @@ class Actor
|
|||
$this->atprotocol = $atprotocol;
|
||||
}
|
||||
|
||||
public function syncContacts(int $uid)
|
||||
/**
|
||||
* Syncronize the contacts (followers, sharers) for the given user
|
||||
*
|
||||
* @param integer $uid User ID
|
||||
* @return void
|
||||
*/
|
||||
public function syncContacts(int $uid): void
|
||||
{
|
||||
$this->logger->info('Sync contacts for user - start', ['uid' => $uid]);
|
||||
$contacts = Contact::selectToArray(['id', 'url', 'rel'], ['uid' => $uid, 'network' => Protocol::BLUESKY, 'rel' => [Contact::FRIEND, Contact::SHARING, Contact::FOLLOWER]]);
|
||||
|
|
@ -93,9 +99,16 @@ class Actor
|
|||
$this->logger->info('Sync contacts for user - done', ['uid' => $uid]);
|
||||
}
|
||||
|
||||
public function updateContactByDID(string $did)
|
||||
/**
|
||||
* Update a contact for a given DID and user id
|
||||
*
|
||||
* @param string $did DID (did:plc:...)
|
||||
* @param integer $contact_uid User id of the contact to be updated
|
||||
* @return void
|
||||
*/
|
||||
public function updateContactByDID(string $did, int $contact_uid): void
|
||||
{
|
||||
$profile = $this->atprotocol->XRPCGet('app.bsky.actor.getProfile', ['actor' => $did]);
|
||||
$profile = $this->atprotocol->XRPCGet('app.bsky.actor.getProfile', ['actor' => $did], $contact_uid);
|
||||
if (empty($profile) || empty($profile->did)) {
|
||||
return;
|
||||
}
|
||||
|
|
@ -139,20 +152,7 @@ class Actor
|
|||
}
|
||||
}
|
||||
|
||||
/*
|
||||
@todo Add this part when the function will be callable with a uid
|
||||
if (!empty($profile->viewer)) {
|
||||
if (!empty($profile->viewer->following) && !empty($profile->viewer->followedBy)) {
|
||||
$fields['rel'] = Contact::FRIEND;
|
||||
} elseif (!empty($profile->viewer->following) && empty($profile->viewer->followedBy)) {
|
||||
$fields['rel'] = Contact::SHARING;
|
||||
} elseif (empty($profile->viewer->following) && !empty($profile->viewer->followedBy)) {
|
||||
$fields['rel'] = Contact::FOLLOWER;
|
||||
} else {
|
||||
$fields['rel'] = Contact::NOTHING;
|
||||
}
|
||||
}
|
||||
*/
|
||||
Contact::update($fields, ['nurl' => $profile->did, 'network' => Protocol::BLUESKY]);
|
||||
|
||||
if (!empty($profile->avatar)) {
|
||||
$contact = Contact::selectFirst(['id', 'avatar'], ['network' => Protocol::BLUESKY, 'nurl' => $did, 'uid' => 0]);
|
||||
|
|
@ -161,16 +161,37 @@ class Actor
|
|||
}
|
||||
}
|
||||
|
||||
$this->logger->notice('Update profile', ['did' => $profile->did, 'fields' => $fields]);
|
||||
$this->logger->notice('Update global profile', ['did' => $profile->did, 'fields' => $fields]);
|
||||
|
||||
Contact::update($fields, ['nurl' => $profile->did, 'network' => Protocol::BLUESKY]);
|
||||
if (!empty($profile->viewer) && ($contact_uid != 0)) {
|
||||
if (!empty($profile->viewer->following) && !empty($profile->viewer->followedBy)) {
|
||||
$user_fields = ['rel' => Contact::FRIEND];
|
||||
} elseif (!empty($profile->viewer->following) && empty($profile->viewer->followedBy)) {
|
||||
$user_fields = ['rel' => Contact::SHARING];
|
||||
} elseif (empty($profile->viewer->following) && !empty($profile->viewer->followedBy)) {
|
||||
$user_fields = ['rel' => Contact::FOLLOWER];
|
||||
} else {
|
||||
$user_fields = ['rel' => Contact::NOTHING];
|
||||
}
|
||||
Contact::update($user_fields, ['nurl' => $profile->did, 'network' => Protocol::BLUESKY, 'uid' => $contact_uid]);
|
||||
$this->logger->notice('Update user profile', ['uid' => $contact_uid, 'did' => $profile->did, 'fields' => $user_fields]);
|
||||
}
|
||||
}
|
||||
|
||||
public function getContactByDID(string $did, int $uid, int $contact_uid): array
|
||||
/**
|
||||
* Fetch and possibly create a contact array for a given DID
|
||||
*
|
||||
* @param string $did The contact DID
|
||||
* @param integer $uid "0" when either the public contact or the user contact is desired
|
||||
* @param integer $contact_uid If not found, the contact will be created for this user id
|
||||
* @param boolean $auto_update Default "false". If activated, the contact will be updated every 24 hours
|
||||
* @return array Contact array
|
||||
*/
|
||||
public function getContactByDID(string $did, int $uid, int $contact_uid, bool $auto_update = false): array
|
||||
{
|
||||
$contact = Contact::selectFirst([], ['network' => Protocol::BLUESKY, 'nurl' => $did, 'uid' => [$contact_uid, $uid]], ['order' => ['uid' => true]]);
|
||||
|
||||
if (!empty($contact)) {
|
||||
if (!empty($contact) && (!$auto_update || ($contact['updated'] > DateTimeFormat::utc('now -24 hours')))) {
|
||||
return $contact;
|
||||
}
|
||||
|
||||
|
|
@ -193,7 +214,7 @@ class Actor
|
|||
|
||||
$cid = Contact::insert($fields);
|
||||
|
||||
$this->updateContactByDID($did);
|
||||
$this->updateContactByDID($did, $contact_uid);
|
||||
|
||||
return Contact::getById($cid);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,9 +38,9 @@ use stdClass;
|
|||
*/
|
||||
class Jetstream
|
||||
{
|
||||
private $uids = [];
|
||||
private $self = [];
|
||||
private $capped = false;
|
||||
private $uids = [];
|
||||
private $self = [];
|
||||
private $capped = false;
|
||||
|
||||
/** @var LoggerInterface */
|
||||
private $logger;
|
||||
|
|
@ -73,10 +73,12 @@ class Jetstream
|
|||
$this->processor = $processor;
|
||||
}
|
||||
|
||||
// *****************************************
|
||||
// * Listener
|
||||
// *****************************************
|
||||
public function listen()
|
||||
/**
|
||||
* Listen to incoming webstream messages from Jetstream
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
public function listen(): void
|
||||
{
|
||||
$timeout = 300;
|
||||
$timeout_limit = 10;
|
||||
|
|
@ -108,6 +110,7 @@ class Jetstream
|
|||
$timestamp = $data->time_us;
|
||||
$this->route($data);
|
||||
$this->keyValue->set('jetstream_timestamp', $timestamp);
|
||||
$this->incrementMessages();
|
||||
} else {
|
||||
$this->logger->warning('Unexpected return value', ['data' => $data]);
|
||||
break;
|
||||
|
|
@ -135,6 +138,25 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Increment the message counter for the statistics page
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
private function incrementMessages(): void
|
||||
{
|
||||
$packets = (int)($this->keyValue->get('jetstream_messages') ?? 0);
|
||||
if ($packets >= PHP_INT_MAX) {
|
||||
$packets = 0;
|
||||
}
|
||||
$this->keyValue->set('jetstream_messages', $packets + 1);
|
||||
}
|
||||
|
||||
/**
|
||||
* Synchronize contacts for all active users
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
private function syncContacts()
|
||||
{
|
||||
$active_uids = $this->atprotocol->getUids();
|
||||
|
|
@ -147,6 +169,11 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set options like the followed DIDs
|
||||
*
|
||||
* @return void
|
||||
*/
|
||||
private function setOptions()
|
||||
{
|
||||
$active_uids = $this->atprotocol->getUids();
|
||||
|
|
@ -184,10 +211,14 @@ class Jetstream
|
|||
}
|
||||
|
||||
if (!$this->capped && count($dids) < $did_limit) {
|
||||
$contacts = Contact::selectToArray(['url'], ['uid' => 0, 'network' => Protocol::BLUESKY], ['order' => ['last-item' => true], 'limit' => $did_limit]);
|
||||
$condition = ["`uid` = ? AND `network` = ? AND EXISTS(SELECT `author-id` FROM `post-user` WHERE `author-id` = `contact`.`id` AND `post-user`.`uid` != ?)", 0, Protocol::BLUESKY, 0];
|
||||
$contacts = Contact::selectToArray(['url'], $condition, ['order' => ['last-item' => true], 'limit' => $did_limit]);
|
||||
$dids = $this->addDids($contacts, $uids, $did_limit, $dids);
|
||||
}
|
||||
|
||||
$this->keyValue->set('jetstream_did_count', count($dids));
|
||||
$this->keyValue->set('jetstream_did_limit', $did_limit);
|
||||
|
||||
$this->logger->debug('Selected DIDs', ['uids' => $active_uids, 'count' => count($dids), 'capped' => $this->capped]);
|
||||
$update = [
|
||||
'type' => 'options_update',
|
||||
|
|
@ -204,6 +235,15 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns an array of DIDs provided by an array of contacts
|
||||
*
|
||||
* @param array $contacts Array of contact records
|
||||
* @param array $uids Array with the user ids with enabled bluesky timeline import
|
||||
* @param integer $did_limit Maximum limit of entries
|
||||
* @param array $dids Array of DIDs that are added to the output list
|
||||
* @return array DIDs
|
||||
*/
|
||||
private function addDids(array $contacts, array $uids, int $did_limit, array $dids): array
|
||||
{
|
||||
foreach ($contacts as $contact) {
|
||||
|
|
@ -218,7 +258,13 @@ class Jetstream
|
|||
return $dids;
|
||||
}
|
||||
|
||||
private function route(stdClass $data)
|
||||
/**
|
||||
* Route incoming messages
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function route(stdClass $data): void
|
||||
{
|
||||
Item::incrementInbound(Protocol::BLUESKY);
|
||||
|
||||
|
|
@ -239,19 +285,15 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function routeCommits(stdClass $data)
|
||||
/**
|
||||
* Route incoming commit messages
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function routeCommits(stdClass $data): void
|
||||
{
|
||||
$drift = max(0, round(time() - $data->time_us / 1000000));
|
||||
if ($drift > 60 && !$this->capped) {
|
||||
$this->capped = true;
|
||||
$this->setOptions();
|
||||
$this->logger->notice('Drift is too high, dids will be capped');
|
||||
} elseif ($drift == 0 && $this->capped) {
|
||||
$this->capped = false;
|
||||
$this->setOptions();
|
||||
$this->logger->notice('Drift is low enough, dids will be uncapped');
|
||||
}
|
||||
|
||||
$drift = $this->getDrift($data);
|
||||
$this->logger->notice('Received commit', ['time' => date(DateTimeFormat::ATOM, $data->time_us / 1000000), 'drift' => $drift, 'capped' => $this->capped, 'did' => $data->did, 'operation' => $data->commit->operation, 'collection' => $data->commit->collection, 'timestamp' => $data->time_us]);
|
||||
$timestamp = microtime(true);
|
||||
|
||||
|
|
@ -295,11 +337,41 @@ class Jetstream
|
|||
break;
|
||||
}
|
||||
if (microtime(true) - $timestamp > 2) {
|
||||
$this->logger->notice('Commit processed', ['duration' => round(microtime(true) - $timestamp, 3), 'time' => date(DateTimeFormat::ATOM, $data->time_us / 1000000), 'did' => $data->did, 'operation' => $data->commit->operation, 'collection' => $data->commit->collection]);
|
||||
$this->logger->notice('Commit processed', ['duration' => round(microtime(true) - $timestamp, 3), 'drift' => $drift, 'capped' => $this->capped, 'time' => date(DateTimeFormat::ATOM, $data->time_us / 1000000), 'did' => $data->did, 'operation' => $data->commit->operation, 'collection' => $data->commit->collection]);
|
||||
}
|
||||
}
|
||||
|
||||
private function routePost(stdClass $data, int $drift)
|
||||
/**
|
||||
* Calculate the drift between the server timestamp and the current time.
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return integer The calculated drift
|
||||
*/
|
||||
private function getDrift(stdClass $data): int
|
||||
{
|
||||
$drift = max(0, round(time() - $data->time_us / 1000000));
|
||||
$this->keyValue->set('jetstream_drift', $drift);
|
||||
|
||||
if ($drift > 60 && !$this->capped) {
|
||||
$this->capped = true;
|
||||
$this->setOptions();
|
||||
$this->logger->notice('Drift is too high, dids will be capped');
|
||||
} elseif ($drift == 0 && $this->capped) {
|
||||
$this->capped = false;
|
||||
$this->setOptions();
|
||||
$this->logger->notice('Drift is low enough, dids will be uncapped');
|
||||
}
|
||||
return $drift;
|
||||
}
|
||||
|
||||
/**
|
||||
* Route app.bsky.feed.post commits
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @param integer $drift
|
||||
* @return void
|
||||
*/
|
||||
private function routePost(stdClass $data, int $drift): void
|
||||
{
|
||||
switch ($data->commit->operation) {
|
||||
case 'delete':
|
||||
|
|
@ -316,7 +388,14 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function routeRepost(stdClass $data, int $drift)
|
||||
/**
|
||||
* Route app.bsky.feed.repost commits
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @param integer $drift
|
||||
* @return void
|
||||
*/
|
||||
private function routeRepost(stdClass $data, int $drift): void
|
||||
{
|
||||
switch ($data->commit->operation) {
|
||||
case 'delete':
|
||||
|
|
@ -333,7 +412,13 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function routeLike(stdClass $data)
|
||||
/**
|
||||
* Route app.bsky.feed.like commits
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function routeLike(stdClass $data): void
|
||||
{
|
||||
switch ($data->commit->operation) {
|
||||
case 'delete':
|
||||
|
|
@ -350,7 +435,13 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function routeProfile(stdClass $data)
|
||||
/**
|
||||
* Route app.bsky.actor.profile commits
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function routeProfile(stdClass $data): void
|
||||
{
|
||||
switch ($data->commit->operation) {
|
||||
case 'delete':
|
||||
|
|
@ -358,11 +449,11 @@ class Jetstream
|
|||
break;
|
||||
|
||||
case 'create':
|
||||
$this->actor->updateContactByDID($data->did);
|
||||
$this->actor->updateContactByDID($data->did, 0);
|
||||
break;
|
||||
|
||||
case 'update':
|
||||
$this->actor->updateContactByDID($data->did);
|
||||
$this->actor->updateContactByDID($data->did, 0);
|
||||
break;
|
||||
|
||||
default:
|
||||
|
|
@ -371,7 +462,13 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function routeFollow(stdClass $data)
|
||||
/**
|
||||
* Route app.bsky.graph.follow commits
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function routeFollow(stdClass $data): void
|
||||
{
|
||||
switch ($data->commit->operation) {
|
||||
case 'delete':
|
||||
|
|
@ -394,7 +491,13 @@ class Jetstream
|
|||
}
|
||||
}
|
||||
|
||||
private function storeCommitMessage(stdClass $data)
|
||||
/**
|
||||
* Store commit messages for debugging purposes
|
||||
*
|
||||
* @param stdClass $data message object
|
||||
* @return void
|
||||
*/
|
||||
private function storeCommitMessage(stdClass $data): void
|
||||
{
|
||||
if ($this->config->get('debug', 'jetstream_log')) {
|
||||
$tempfile = tempnam(System::getTempPath(), 'at-proto.commit.' . $data->commit->collection . '.' . $data->commit->operation . '-');
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ namespace Friendica\Protocol\ATProtocol;
|
|||
use Friendica\App\BaseURL;
|
||||
use Friendica\Core\Protocol;
|
||||
use Friendica\Database\Database;
|
||||
use Friendica\Database\DBA;
|
||||
use Friendica\Model\Contact;
|
||||
use Friendica\Model\Conversation;
|
||||
use Friendica\Model\Item;
|
||||
|
|
@ -130,12 +131,12 @@ class Processor
|
|||
if (!empty($data->commit->record->reply)) {
|
||||
$root = $this->getUri($data->commit->record->reply->root);
|
||||
$parent = $this->getUri($data->commit->record->reply->parent);
|
||||
$uids = $this->getPostUids($root);
|
||||
$uids = $this->getPostUids($root, true);
|
||||
if (!$uids) {
|
||||
$this->logger->debug('Comment is not imported since the root post is not found.', ['root' => $root, 'parent' => $parent]);
|
||||
return;
|
||||
}
|
||||
if ($dont_fetch && !$this->getPostUids($parent)) {
|
||||
if ($dont_fetch && !$this->getPostUids($parent, false)) {
|
||||
$this->logger->debug('Comment is not imported since the parent post is not found.', ['root' => $root, 'parent' => $parent]);
|
||||
return;
|
||||
}
|
||||
|
|
@ -168,6 +169,7 @@ class Processor
|
|||
return;
|
||||
}
|
||||
}
|
||||
$item['source'] = json_encode($post);
|
||||
$item = $this->addMedia($post->thread->post->embed, $item, 0);
|
||||
}
|
||||
|
||||
|
|
@ -185,7 +187,7 @@ class Processor
|
|||
|
||||
public function createRepost(stdClass $data, array $uids, bool $dont_fetch)
|
||||
{
|
||||
if ($dont_fetch && !$this->getPostUids($this->getUri($data->commit->record->subject))) {
|
||||
if ($dont_fetch && !$this->getPostUids($this->getUri($data->commit->record->subject), true)) {
|
||||
$this->logger->debug('Repost is not imported since the subject is not found.', ['subject' => $this->getUri($data->commit->record->subject)]);
|
||||
return;
|
||||
}
|
||||
|
|
@ -215,7 +217,7 @@ class Processor
|
|||
|
||||
public function createLike(stdClass $data)
|
||||
{
|
||||
$uids = $this->getPostUids($this->getUri($data->commit->record->subject));
|
||||
$uids = $this->getPostUids($this->getUri($data->commit->record->subject), false);
|
||||
if (!$uids) {
|
||||
$this->logger->debug('Like is not imported since the subject is not found.', ['subject' => $this->getUri($data->commit->record->subject)]);
|
||||
return;
|
||||
|
|
@ -272,7 +274,7 @@ class Processor
|
|||
return true;
|
||||
}
|
||||
|
||||
private function processPost(stdClass $post, int $uid, int $post_reason, int $causer, int $level, int $protocol): int
|
||||
public function processPost(stdClass $post, int $uid, int $post_reason, int $causer, int $level, int $protocol): int
|
||||
{
|
||||
$uri = $this->getUri($post);
|
||||
|
||||
|
|
@ -380,7 +382,7 @@ class Processor
|
|||
return $item;
|
||||
}
|
||||
|
||||
private function getHeaderFromPost(stdClass $post, string $uri, int $uid, int $protocol): array
|
||||
public function getHeaderFromPost(stdClass $post, string $uri, int $uid, int $protocol): array
|
||||
{
|
||||
$parts = $this->getUriParts($uri);
|
||||
if (empty($post->author) || empty($post->cid) || empty($parts->rkey)) {
|
||||
|
|
@ -540,6 +542,8 @@ class Processor
|
|||
'url' => $image->fullsize,
|
||||
'preview' => $image->thumb,
|
||||
'description' => $image->alt,
|
||||
'height' => $image->aspectRatio->height ?? null,
|
||||
'width' => $image->aspectRatio->width ?? null,
|
||||
];
|
||||
Post\Media::insert($media);
|
||||
}
|
||||
|
|
@ -563,6 +567,7 @@ class Processor
|
|||
'uri-id' => $item['uri-id'],
|
||||
'type' => Post\Media::HTML,
|
||||
'url' => $embed->external->uri,
|
||||
'preview' => $embed->external->thumb ?? null,
|
||||
'name' => $embed->external->title,
|
||||
'description' => $embed->external->description,
|
||||
];
|
||||
|
|
@ -688,7 +693,7 @@ class Processor
|
|||
return $restrict ? Item::CANT_REPLY : null;
|
||||
}
|
||||
|
||||
private function fetchMissingPost(string $uri, int $uid, int $post_reason, int $causer, int $level, string $fallback = '', bool $always_fetch = false, int $Protocol = Conversation::PARCEL_JETSTREAM): string
|
||||
public function fetchMissingPost(string $uri, int $uid, int $post_reason, int $causer, int $level, string $fallback = '', bool $always_fetch = false, int $Protocol = Conversation::PARCEL_JETSTREAM): string
|
||||
{
|
||||
$timestamp = microtime(true);
|
||||
$stamp = Strings::getRandomHex(30);
|
||||
|
|
@ -796,7 +801,7 @@ class Processor
|
|||
return $uri;
|
||||
}
|
||||
|
||||
private function getUriParts(string $uri): ?stdClass
|
||||
public function getUriParts(string $uri): ?stdClass
|
||||
{
|
||||
$class = $this->getUriClass($uri);
|
||||
if (empty($class)) {
|
||||
|
|
@ -814,7 +819,7 @@ class Processor
|
|||
return $class;
|
||||
}
|
||||
|
||||
private function getUriClass(string $uri): ?stdClass
|
||||
public function getUriClass(string $uri): ?stdClass
|
||||
{
|
||||
if (empty($uri)) {
|
||||
return null;
|
||||
|
|
@ -839,7 +844,7 @@ class Processor
|
|||
return $class;
|
||||
}
|
||||
|
||||
private function fetchUriId(string $uri, int $uid): string
|
||||
public function fetchUriId(string $uri, int $uid): string
|
||||
{
|
||||
$reply = Post::selectFirst(['uri-id'], ['uri' => $uri, 'uid' => [$uid, 0]]);
|
||||
if (!empty($reply['uri-id'])) {
|
||||
|
|
@ -854,16 +859,18 @@ class Processor
|
|||
return 0;
|
||||
}
|
||||
|
||||
private function getPostUids(string $uri): array
|
||||
private function getPostUids(string $uri, bool $with_public_user): array
|
||||
{
|
||||
$condition = $with_public_user ? [] : ["`uid` != ?", 0];
|
||||
|
||||
$uids = [];
|
||||
$posts = Post::select(['uid'], ['uri' => $uri]);
|
||||
$posts = Post::select(['uid'], DBA::mergeConditions(['uri' => $uri], $condition));
|
||||
while ($post = Post::fetch($posts)) {
|
||||
$uids[] = $post['uid'];
|
||||
}
|
||||
$this->db->close($posts);
|
||||
|
||||
$posts = Post::select(['uid'], ['extid' => $uri]);
|
||||
$posts = Post::select(['uid'], DBA::mergeConditions(['extid' => $uri], $condition));
|
||||
while ($post = Post::fetch($posts)) {
|
||||
$uids[] = $post['uid'];
|
||||
}
|
||||
|
|
@ -880,7 +887,7 @@ class Processor
|
|||
return Post::exists(['extid' => $uri, 'uid' => $uids]);
|
||||
}
|
||||
|
||||
private function getUri(stdClass $post): string
|
||||
public function getUri(stdClass $post): string
|
||||
{
|
||||
if (empty($post->cid)) {
|
||||
$this->logger->info('Invalid URI', ['post' => $post]);
|
||||
|
|
@ -889,7 +896,7 @@ class Processor
|
|||
return $post->uri . ':' . $post->cid;
|
||||
}
|
||||
|
||||
private function getPostUri(string $uri, int $uid): string
|
||||
public function getPostUri(string $uri, int $uid): string
|
||||
{
|
||||
if (Post::exists(['uri' => $uri, 'uid' => [$uid, 0]])) {
|
||||
$this->logger->debug('Post exists', ['uri' => $uri]);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue