Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
93.48%
215 / 230
71.43%
25 / 35
 CRAP
0.00%
0 / 1
OneRecordClient
93.48%
215 / 230
71.43%
25 / 35
 123.99
0.00%
0 / 1
 __construct
100.00%
9 / 9
100.00%
1 / 1
 3
 endpoint
0.00%
0 / 1
0.00%
0 / 1
 2
 withApiVersion
100.00%
3 / 3
100.00%
1 / 1
 1
 serverInformation
87.50%
21 / 24
0.00%
0 / 1
 13.33
 apiVersion
100.00%
8 / 8
100.00%
1 / 1
 4
 getLogisticsObject
100.00%
18 / 18
100.00%
1 / 1
 8
 headLogisticsObject
100.00%
1 / 1
100.00%
1 / 1
 1
 createLogisticsObject
100.00%
2 / 2
100.00%
1 / 1
 2
 requestChange
100.00%
1 / 1
100.00%
1 / 1
 1
 requestVerification
100.00%
1 / 1
100.00%
1 / 1
 1
 getAuditTrail
87.50%
14 / 16
0.00%
0 / 1
 7.10
 postLogisticsEvent
100.00%
3 / 3
100.00%
1 / 1
 2
 postLogisticsEvents
93.75%
15 / 16
0.00%
0 / 1
 6.01
 getLogisticsEvents
92.31%
12 / 13
0.00%
0 / 1
 4.01
 getLogisticsEvent
100.00%
4 / 4
100.00%
1 / 1
 1
 subscribe
100.00%
1 / 1
100.00%
1 / 1
 1
 getSubscriptions
100.00%
11 / 11
100.00%
1 / 1
 3
 requestAccessDelegation
100.00%
4 / 4
100.00%
1 / 1
 3
 sendNotification
100.00%
1 / 1
100.00%
1 / 1
 2
 getActionRequest
100.00%
2 / 2
100.00%
1 / 1
 1
 updateActionRequestStatus
100.00%
1 / 1
100.00%
1 / 1
 1
 revokeActionRequest
100.00%
1 / 1
100.00%
1 / 1
 1
 send
92.31%
24 / 26
0.00%
0 / 1
 13.08
 parse
60.00%
3 / 5
0.00%
0 / 1
 3.58
 objectResponse
100.00%
5 / 5
100.00%
1 / 1
 4
 bulkResults
93.33%
14 / 15
0.00%
0 / 1
 4.00
 event
100.00%
6 / 6
100.00%
1 / 1
 1
 collect
83.33%
5 / 6
0.00%
0 / 1
 7.23
 withoutRevisionProperties
100.00%
6 / 6
100.00%
1 / 1
 4
 originOf
100.00%
8 / 8
100.00%
1 / 1
 7
 iri
100.00%
1 / 1
100.00%
1 / 1
 2
 body
100.00%
1 / 1
100.00%
1 / 1
 1
 location
75.00%
3 / 4
0.00%
0 / 1
 2.06
 httpDate
100.00%
5 / 5
100.00%
1 / 1
 3
 versionParameter
100.00%
1 / 1
100.00%
1 / 1
 2
1<?php
2
3declare(strict_types=1);
4
5namespace LambdaTwelve\OneRecord\Client;
6
7use DateTimeImmutable;
8use DateTimeZone;
9use InvalidArgumentException;
10use LambdaTwelve\OneRecord\Api\AccessDelegation;
11use LambdaTwelve\OneRecord\Api\ActionRequest;
12use LambdaTwelve\OneRecord\Api\ErrorDocument;
13use LambdaTwelve\OneRecord\Api\InvalidDocument;
14use LambdaTwelve\OneRecord\Api\Notification;
15use LambdaTwelve\OneRecord\Api\RequestStatus;
16use LambdaTwelve\OneRecord\Api\ServerInformation;
17use LambdaTwelve\OneRecord\Api\Subscription;
18use LambdaTwelve\OneRecord\Api\TopicType;
19use LambdaTwelve\OneRecord\Api\Verification;
20use LambdaTwelve\OneRecord\Change\Change;
21use LambdaTwelve\OneRecord\JsonLd\ExpandedDocument;
22use LambdaTwelve\OneRecord\JsonLd\Json;
23use LambdaTwelve\OneRecord\JsonLd\JsonLd;
24use LambdaTwelve\OneRecord\JsonLd\JsonLdException;
25use LambdaTwelve\OneRecord\JsonLd\Nodes;
26use LambdaTwelve\OneRecord\Model\LogisticsEvent;
27use LambdaTwelve\OneRecord\Model\LogisticsObject;
28use LambdaTwelve\OneRecord\Model\ModelException;
29use LambdaTwelve\OneRecord\Rdf\BlankNode;
30use LambdaTwelve\OneRecord\Rdf\Graph;
31use LambdaTwelve\OneRecord\Rdf\Iri;
32use LambdaTwelve\OneRecord\Rdf\Triple;
33use LambdaTwelve\OneRecord\Spec\ApiFeatures;
34use LambdaTwelve\OneRecord\Spec\ApiVersion;
35use LambdaTwelve\OneRecord\Vocabulary\Generated\Api;
36use LambdaTwelve\OneRecord\Vocabulary\Generated\Cargo;
37use Psr\Clock\ClockInterface;
38use Psr\Http\Client\ClientExceptionInterface;
39use Psr\Http\Client\ClientInterface;
40use Psr\Http\Message\RequestFactoryInterface;
41use Psr\Http\Message\ResponseInterface;
42use Psr\Http\Message\StreamFactoryInterface;
43use Psr\Log\LoggerInterface;
44use Psr\Log\NullLogger;
45use Psr\SimpleCache\CacheInterface;
46use Throwable;
47
48/**
49 * A client for one partner's ONE Record server over PSR-18. It discovers the
50 * partner's server information, picks the highest API version both sides
51 * support and speaks that version from then on. Responses are parsed into the
52 * same model and API documents the server side uses; HTTP errors become
53 * OneRecordHttpException with the api:Error the partner sent.
54 *
55 * Parsing is lenient on purpose: unknown predicates are kept in the graph
56 * and 2.3-only properties are optional, so a 2.2 partner is read fine.
57 */
58final class OneRecordClient
59{
60    private const string JSON_LD = 'application/ld+json';
61
62    private readonly string $endpoint;
63
64    /** @var list<string> normalised origins (scheme://host:port) a request may be sent to with this partner's token */
65    private readonly array $origins;
66
67    /** @var list<ApiVersion> */
68    private readonly array $ourVersions;
69
70    private ?ApiVersion $version = null;
71
72    private ?ServerInformation $serverInformation = null;
73
74    private readonly LoggerInterface $logger;
75
76    /**
77     * @param string $serverEndpoint the partner's ONE Record server endpoint (what its server information calls api:hasServerEndpoint)
78     * @param ?list<ApiVersion> $apiVersions the versions this client is willing to speak; all supported ones by default
79     * @param int $serverInformationTtl seconds to keep a partner's server information in the cache
80     * @param list<string> $additionalOrigins other origins (scheme://host[:port]) this partner legitimately serves
81     *                                        resources from; the token is sent there too. Empty for almost everyone.
82     */
83    public function __construct(
84        private readonly ClientInterface $http,
85        private readonly RequestFactoryInterface $requests,
86        private readonly StreamFactoryInterface $streams,
87        private readonly TokenProvider $tokens,
88        string $serverEndpoint,
89        private readonly ?CacheInterface $cache = null,
90        private readonly ?ClockInterface $clock = null,
91        ?LoggerInterface $logger = null,
92        ?array $apiVersions = null,
93        private readonly int $serverInformationTtl = 3600,
94        array $additionalOrigins = [],
95    ) {
96        if (preg_match('#^https?://#', $serverEndpoint) !== 1) {
97            throw new InvalidArgumentException(\sprintf('The server endpoint must be an absolute http(s) URL, got "%s".', $serverEndpoint));
98        }
99        $this->endpoint = rtrim($serverEndpoint, '/');
100        $origins = [self::originOf($this->endpoint) ?? throw new InvalidArgumentException(\sprintf('The server endpoint "%s" has no usable origin.', $serverEndpoint))];
101        foreach ($additionalOrigins as $origin) {
102            $origins[] = self::originOf($origin) ?? throw new InvalidArgumentException(\sprintf('"%s" is not an origin (scheme://host[:port]).', $origin));
103        }
104        $this->origins = $origins;
105        $this->ourVersions = $apiVersions ?? ApiVersion::allDescending();
106        $this->logger = $logger ?? new NullLogger();
107    }
108
109    public function endpoint(): string
110    {
111        return $this->endpoint;
112    }
113
114    /**
115     * Speak this version regardless of what the partner advertises.
116     */
117    public function withApiVersion(ApiVersion $version): self
118    {
119        $clone = clone $this;
120        $clone->version = $version;
121
122        return $clone;
123    }
124
125    /**
126     * The partner's server information, cached (PSR-16 when given, else in this object).
127     */
128    public function serverInformation(bool $refresh = false): ServerInformation
129    {
130        if (!$refresh && $this->serverInformation !== null) {
131            return $this->serverInformation;
132        }
133        $key = 'one-record.server-information.' . hash('sha256', $this->endpoint);
134        if (!$refresh && $this->cache !== null) {
135            try {
136                $cached = $this->cache->get($key);
137                if (\is_string($cached)) {
138                    return $this->serverInformation = ServerInformation::fromJsonLd($cached);
139                }
140            } catch (Throwable) {
141                // A cache problem only costs a request.
142            }
143        }
144        // Server information is fetched before a version is negotiated. A strict partner answers 406
145        // to a version it does not speak, so ask for each of ours in turn, highest first (AR-005).
146        $response = null;
147        $refused = null;
148        foreach ($this->ourVersions as $candidate) {
149            try {
150                $response = $this->send('GET', $this->endpoint . '/', null, $candidate);
151                break;
152            } catch (OneRecordHttpException $e) {
153                if ($e->status !== 406) {
154                    throw $e;
155                }
156                $refused = $e;
157            }
158        }
159        if ($response === null) {
160            throw new ClientException(\sprintf('%s accepts none of the API versions this client speaks (%s).', $this->endpoint, implode(', ', array_map(static fn(ApiVersion $v): string => $v->value, $this->ourVersions))), 0, $refused);
161        }
162        $information = $this->parse(static fn(): ServerInformation => ServerInformation::fromJsonLd(self::body($response)), 'server information');
163        if ($this->cache !== null) {
164            try {
165                $this->cache->set($key, json_encode($information->toJsonLd(), JSON_THROW_ON_ERROR), $this->serverInformationTtl);
166            } catch (Throwable) {
167            }
168        }
169
170        return $this->serverInformation = $information;
171    }
172
173    /**
174     * The API version used with this partner: the highest both sides support.
175     *
176     * @throws ClientException when the partner supports none of ours
177     */
178    public function apiVersion(): ApiVersion
179    {
180        if ($this->version !== null) {
181            return $this->version;
182        }
183        $information = $this->serverInformation();
184        $common = $information->bestCommonApiVersion($this->ourVersions);
185        if ($common === null) {
186            throw new ClientException(\sprintf('%s supports API %s; this client speaks %s.', $this->endpoint, $information->apiVersions === [] ? 'no listed version' : implode(', ', $information->apiVersions), implode(', ', array_map(static fn(ApiVersion $v): string => $v->value, $this->ourVersions))));
187        }
188        $this->logger->debug('ONE Record API version negotiated', ['endpoint' => $this->endpoint, 'version' => $common->value]);
189
190        return $this->version = $common;
191    }
192
193    // --- Logistics objects -------------------------------------------------
194
195    /**
196     * @param ?DateTimeImmutable $at the revision current at that instant (the spec's `?at=` parameter)
197     * @param bool $embedded ask the server to embed linked objects it holds
198     */
199    public function getLogisticsObject(Iri|string $object, ?DateTimeImmutable $at = null, bool $embedded = false): LogisticsObjectResponse
200    {
201        $iri = self::iri($object);
202        $query = [];
203        if ($at !== null) {
204            $query['at'] = $at->setTimezone(new DateTimeZone('UTC'))->format('Ymd\THis\Z');
205        }
206        if ($embedded) {
207            $query['embedded'] = 'true';
208        }
209        $url = $iri->value . ($query === [] ? '' : '?' . http_build_query($query, '', '&', PHP_QUERY_RFC3986));
210        $response = $this->send('GET', $url);
211        $object = $this->parse(static function () use ($response, $iri, $query): LogisticsObject {
212            // The body must be about the object asked for. A historical read may be rooted at
213            // the "<iri>?at=…" URL instead; nothing else is accepted (AR-017). Both identities are
214            // named before the root is chosen, so a flattened answer is rooted at one of them wherever
215            // the node sits (R3-004).
216            $allowed = [$iri->value];
217            if (isset($query['at'])) {
218                // The historical identity carries the instant only; embedded=true is presentation, not identity (R2-003).
219                $allowed[] = $iri->value . '?at=' . $query['at'];
220            }
221            $document = JsonLd::expand(self::body($response), array_map(static fn(string $a): Iri => new Iri($a), $allowed));
222            if (!$document->root instanceof Iri || !\in_array($document->root->value, $allowed, true)) {
223                throw new ClientException(\sprintf('Asked for %s, the server answered with a document about %s.', $iri->value, $document->root instanceof Iri ? $document->root->value : 'an unidentified node'));
224            }
225
226            return LogisticsObject::fromJsonLd(self::body($response), $document->root);
227        }, 'logistics object');
228
229        return $this->objectResponse($response, self::withoutRevisionProperties($object));
230    }
231
232    public function headLogisticsObject(Iri|string $object): LogisticsObjectResponse
233    {
234        return $this->objectResponse($this->send('HEAD', self::iri($object)->value), null);
235    }
236
237    /**
238     * POST /logistics-objects on the partner's server (it must allow it); returns the URI assigned.
239     *
240     * @param LogisticsObject|array<string, mixed> $object
241     */
242    public function createLogisticsObject(LogisticsObject|array $object): Iri
243    {
244        $json = $object instanceof LogisticsObject ? $object->toJsonLd() : $object;
245
246        return self::location($this->send('POST', $this->endpoint . '/logistics-objects', $json));
247    }
248
249    /**
250     * PATCH a Change; returns the URI of the ChangeRequest the partner created.
251     */
252    public function requestChange(Change $change): Iri
253    {
254        return self::location($this->send('PATCH', $change->logisticsObject->value, $change->toJsonLd()));
255    }
256
257    /**
258     * POST a Verification on the object; returns the URI of the VerificationRequest.
259     */
260    public function requestVerification(Verification $verification): Iri
261    {
262        return self::location($this->send('POST', $verification->logisticsObject->value, $verification->toJsonLd($this->apiVersion())));
263    }
264
265    public function getAuditTrail(Iri|string $object, ?DateTimeImmutable $updatedFrom = null, ?DateTimeImmutable $updatedTo = null, ?RequestStatus $status = null): AuditTrail
266    {
267        $query = [];
268        foreach (['updated-from' => $updatedFrom, 'updated-to' => $updatedTo] as $name => $value) {
269            if ($value !== null) {
270                $query[$name] = $value->setTimezone(new DateTimeZone('UTC'))->format('Ymd\THis\Z');
271            }
272        }
273        if ($status !== null) {
274            $query['status'] = $status->shortName();
275        }
276        $response = $this->send('GET', self::iri($object)->value . '/audit-trail' . ($query === [] ? '' : '?' . http_build_query($query, '', '&', PHP_QUERY_RFC3986)));
277
278        return $this->parse(function () use ($response): AuditTrail {
279            $expanded = JsonLd::expand(self::body($response));
280            if (!\in_array(Api::AuditTrail, $expanded->rootTypes(), true)) {
281                throw InvalidDocument::because('Invalid resource', 'The body is not an api:AuditTrail.');
282            }
283            $requests = [];
284            foreach (Nodes::nodes($expanded->graph, $expanded->root, Api::hasActionRequest) as $node) {
285                $requests[] = ActionRequest::fromJsonLd(new ExpandedDocument($expanded->graph, $node, $expanded->context));
286            }
287
288            return new AuditTrail(Nodes::int($expanded->graph, $expanded->root, Api::hasLatestRevision) ?? 0, $requests, self::httpDate($response, 'Last-Modified'));
289        }, 'audit trail');
290    }
291
292    // --- Logistics events --------------------------------------------------
293
294    /**
295     * POST one event on one object; returns the event's URI.
296     *
297     * @param LogisticsEvent|array<string, mixed> $event
298     */
299    public function postLogisticsEvent(Iri|string $object, LogisticsEvent|array $event): Iri
300    {
301        $json = $event instanceof LogisticsEvent ? $event->toJsonLd() : $event;
302        unset($json['@id']);
303
304        return self::location($this->send('POST', self::iri($object)->value . '/logistics-events', $json));
305    }
306
307    /**
308     * One event for several objects. Uses the 2.3 bulk endpoint when the
309     * negotiated version has it and the partner serves it; otherwise posts to
310     * each object in turn. Either way the answer is one result per object.
311     *
312     * @param array<string, mixed> $event the event without cargo:eventFor
313     * @param list<Iri|string> $objects
314     * @return list<BulkEventResult>
315     */
316    public function postLogisticsEvents(array $event, array $objects): array
317    {
318        $iris = array_map(self::iri(...), $objects);
319        unset($event['@id'], $event['cargo:eventFor'], $event[Cargo::eventFor]);
320        if (ApiFeatures::available($this->apiVersion(), ApiFeatures::BULK_LOGISTICS_EVENTS)) {
321            try {
322                $body = [...$event, 'cargo:eventFor' => array_map(static fn(Iri $i): array => ['@id' => $i->value], $iris)];
323                $response = $this->send('POST', $this->endpoint . '/logistics-events', $body, expect: [207]);
324
325                return $this->parse(fn(): array => $this->bulkResults($response, $iris), 'multi-status response');
326            } catch (OneRecordHttpException $e) {
327                if (!\in_array($e->status, [404, 405], true)) {
328                    throw $e;
329                }
330                $this->logger->info('Bulk events not served; posting per object', ['endpoint' => $this->endpoint, 'status' => $e->status]);
331            }
332        }
333        $results = [];
334        foreach ($iris as $iri) {
335            try {
336                $results[] = new BulkEventResult($iri, 201, $this->postLogisticsEvent($iri, $event), null);
337            } catch (OneRecordHttpException $e) {
338                $results[] = new BulkEventResult($iri, $e->status, null, $e->error);
339            }
340        }
341
342        return $results;
343    }
344
345    public function getLogisticsEvents(Iri|string $object, ?EventFilter $filter = null): EventList
346    {
347        $iri = self::iri($object);
348        $response = $this->send('GET', $iri->value . '/logistics-events' . ($filter ?? EventFilter::all())->toQueryString());
349
350        return $this->parse(function () use ($response, $iri): EventList {
351            $expanded = JsonLd::expand(self::body($response));
352            if (!\in_array(Api::Collection, $expanded->rootTypes(), true)) {
353                throw InvalidDocument::because('Invalid resource', 'The body is not an api:Collection.');
354            }
355            $lastModified = self::httpDate($response, 'Last-Modified');
356            $events = [];
357            foreach (Nodes::nodes($expanded->graph, $expanded->root, Api::hasItem) as $node) {
358                if ($node instanceof Iri) {
359                    $events[] = $this->event($expanded->graph, $node, $iri, $lastModified);
360                }
361            }
362
363            return new EventList($events, Nodes::int($expanded->graph, $expanded->root, Api::hasTotalItems) ?? \count($events), $lastModified);
364        }, 'event list');
365    }
366
367    public function getLogisticsEvent(Iri|string $event): LogisticsEvent
368    {
369        $iri = self::iri($event);
370        $response = $this->send('GET', $iri->value);
371        $object = new Iri(preg_replace('#/logistics-events/[^/]+$#', '', $iri->value) ?? $iri->value);
372
373        return $this->parse(fn(): LogisticsEvent => $this->event(JsonLd::expand(self::body($response))->graph, $iri, $object, self::httpDate($response, 'Last-Modified')), 'logistics event');
374    }
375
376    // --- Subscriptions, delegations, notifications --------------------------
377
378    /**
379     * POST /subscriptions; returns the URI of the SubscriptionRequest.
380     */
381    public function subscribe(Subscription $subscription): Iri
382    {
383        return self::location($this->send('POST', $this->endpoint . '/subscriptions', $subscription->toJsonLd()));
384    }
385
386    /**
387     * GET /subscriptions?topicType&topic: what the partner wants to be told about a topic.
388     *
389     * @return list<Subscription> none, one, or several (the partner may answer with a Collection)
390     */
391    public function getSubscriptions(TopicType $topicType, string $topic): array
392    {
393        $query = http_build_query(['topicType' => $topicType->shortName(), 'topic' => $topic], '', '&', PHP_QUERY_RFC3986);
394        $response = $this->send('GET', $this->endpoint . '/subscriptions?' . $query);
395
396        return $this->parse(static function () use ($response): array {
397            $expanded = JsonLd::expand(self::body($response));
398            if (\in_array(Api::Subscription, $expanded->rootTypes(), true)) {
399                return [Subscription::readNode($expanded->graph, $expanded->root)];
400            }
401            $out = [];
402            foreach (Nodes::nodes($expanded->graph, $expanded->root, Api::hasItem) as $node) {
403                $out[] = Subscription::readNode($expanded->graph, $node);
404            }
405
406            return $out;
407        }, 'subscriptions');
408    }
409
410    /**
411     * POST /access-delegations; returns the URI of the AccessDelegationRequest.
412     */
413    public function requestAccessDelegation(AccessDelegation $delegation): Iri
414    {
415        $version = $this->apiVersion();
416        if ($delegation->expiresAt !== null && !ApiFeatures::available($version, ApiFeatures::ACCESS_DELEGATION_EXPIRY)) {
417            // Rendering an old edition may drop a property; requesting an authorization may not silently widen it (AR-026).
418            throw new ClientException(\sprintf('%s speaks API %s, which cannot express api:expiresAt; an unlimited delegation would be requested instead. Drop the expiry deliberately or speak to a 2.3 partner.', $this->endpoint, $version->value));
419        }
420
421        return self::location($this->send('POST', $this->endpoint . '/access-delegations', $delegation->toJsonLd($version)));
422    }
423
424    /**
425     * POST /notifications on the partner's server (the partner is the subscriber).
426     */
427    /**
428     * @param ?string $idempotencyKey a stable id for this delivery (OutboundNotification::$id), sent as
429     *                                an Idempotency-Key header so a receiver can drop a retry (AR-024)
430     */
431    public function sendNotification(Notification $notification, ?string $idempotencyKey = null): void
432    {
433        $this->send('POST', $this->endpoint . '/notifications', $notification->toJsonLd(), expect: [204, 200], headers: $idempotencyKey === null ? [] : ['Idempotency-Key' => $idempotencyKey]);
434    }
435
436    // --- Action requests ---------------------------------------------------
437
438    public function getActionRequest(Iri|string $request): ActionRequest
439    {
440        $response = $this->send('GET', self::iri($request)->value);
441
442        return $this->parse(static fn(): ActionRequest => ActionRequest::fromJsonLd(self::body($response)), 'action request');
443    }
444
445    /**
446     * PATCH /action-requests/{id}?status=…: accept, reject, acknowledge or revoke
447     * a request on a partner's server (its policy decides whether we may).
448     */
449    public function updateActionRequestStatus(Iri|string $request, RequestStatus $status): void
450    {
451        $this->send('PATCH', self::iri($request)->value . '?status=' . $status->shortName(), null, expect: [204]);
452    }
453
454    public function revokeActionRequest(Iri|string $request): void
455    {
456        $this->send('DELETE', self::iri($request)->value, null, expect: [204]);
457    }
458
459    // --- Plumbing ----------------------------------------------------------
460
461    /**
462     * @param ?array<string, mixed> $body
463     * @param list<int> $expect acceptable status codes; empty means any 2xx
464     * @param array<string, string> $headers
465     */
466    private function send(string $method, string $url, ?array $body = null, ?ApiVersion $version = null, array $expect = [], array $headers = []): ResponseInterface
467    {
468        // The token is this partner's; a resource IRI pointing anywhere else must not carry it (AR-001).
469        $origin = self::originOf($url);
470        if ($origin === null || !\in_array($origin, $this->origins, true)) {
471            throw new ClientException(\sprintf('%s is not on %s; this client sends its credentials only to that server (pass additionalOrigins for a partner that serves resources from several hosts).', $url, implode(', ', $this->origins)));
472        }
473        $version ??= $this->apiVersion();
474        $request = $this->requests->createRequest($method, $url)
475            ->withHeader('Accept', self::JSON_LD . '; version=' . $version->value)
476            ->withHeader('Authorization', 'Bearer ' . $this->tokens->token($this->endpoint));
477        if ($body !== null) {
478            $request = $request
479                ->withHeader('Content-Type', self::JSON_LD . '; version=' . $version->value)
480                ->withBody($this->streams->createStream(Json::encode($body, false)));
481        }
482        foreach ($headers as $name => $value) {
483            $request = $request->withHeader($name, $value);
484        }
485        try {
486            $response = $this->http->sendRequest($request);
487        } catch (ClientExceptionInterface $e) {
488            throw new ClientException(\sprintf('%s %s failed: %s', $method, $url, $e->getMessage()), 0, $e);
489        }
490        $status = $response->getStatusCode();
491        if ($status >= 400) {
492            $error = null;
493            try {
494                $error = ErrorDocument::read((string) $response->getBody());
495            } catch (Throwable) {
496                // Not a ONE Record error body (a proxy page, say); the status still tells the story.
497            }
498            $this->logger->info('ONE Record request refused', ['method' => $method, 'url' => $url, 'status' => $status, 'title' => $error?->title]);
499
500            throw new OneRecordHttpException($status, $error, $response, $method, $url);
501        }
502        if ($status < 200 || $status >= 300 || ($expect !== [] && !\in_array($status, $expect, true))) {
503            throw new ClientException(\sprintf('%s %s answered %d where %s was expected.', $method, $url, $status, $expect === [] ? 'a 2xx' : implode(' or ', $expect)));
504        }
505
506        return $response;
507    }
508
509    /**
510     * @template T
511     * @param callable(): T $parse
512     * @return T
513     */
514    private function parse(callable $parse, string $what)
515    {
516        try {
517            return $parse();
518        } catch (ClientException $e) {
519            throw $e;
520        } catch (JsonLdException|InvalidDocument|ModelException $e) {
521            throw new ClientException(\sprintf('%s answered with a %s this client cannot read: %s', $this->endpoint, $what, $e->getMessage()), 0, $e);
522        }
523    }
524
525    private function objectResponse(ResponseInterface $response, ?LogisticsObject $object): LogisticsObjectResponse
526    {
527        $revision = (int) $response->getHeaderLine('Revision');
528        $latest = (int) $response->getHeaderLine('Latest-Revision');
529        $type = $response->getHeaderLine('Type');
530        $version = ApiVersion::tryFromString(self::versionParameter($response->getHeaderLine('Content-Type')) ?? '') ?? $this->apiVersion();
531
532        return new LogisticsObjectResponse($object, $revision > 0 ? $revision : $latest, $latest > 0 ? $latest : $revision, self::httpDate($response, 'Last-Modified'), $type === '' ? $object?->mostSpecificType() : $type, $version);
533    }
534
535    /**
536     * @param list<Iri> $objects
537     * @return list<BulkEventResult>
538     */
539    private function bulkResults(ResponseInterface $response, array $objects): array
540    {
541        $expanded = JsonLd::expand(self::body($response));
542        $graph = $expanded->graph;
543        $results = [];
544        foreach (Nodes::nodes($graph, $expanded->root, Api::hasCreationResult) as $node) {
545            $object = Nodes::iri($graph, $node, Api::hasLogisticsObject);
546            if ($object === null) {
547                continue;
548            }
549            $errorNode = Nodes::node($graph, $node, Api::hasError);
550            $results[] = new BulkEventResult(
551                $object,
552                Nodes::int($graph, $node, Api::hasHTTPStatus) ?? 0,
553                Nodes::iri($graph, $node, Api::hasLogisticsEvent),
554                $errorNode === null ? null : ErrorDocument::readNode($graph, $errorNode),
555            );
556        }
557
558        return $results;
559    }
560
561    private function event(Graph $graph, Iri $iri, Iri $object, ?DateTimeImmutable $fallbackCreated): LogisticsEvent
562    {
563        $subgraph = new Graph();
564        $seen = [];
565        self::collect($graph, $iri, $subgraph, $seen);
566        $for = Nodes::iri($subgraph, $iri, Cargo::eventFor);
567        $created = Nodes::dateTime($subgraph, $iri, Cargo::creationDate) ?? $fallbackCreated ?? ($this->clock?->now() ?? new DateTimeImmutable('now', new DateTimeZone('UTC')));
568
569        return new LogisticsEvent($iri, $for ?? $object, $subgraph, $created);
570    }
571
572    /**
573     * One visited set for the whole walk: a set per branch would terminate cycles but
574     * revisit every shared descendant, which is exponential on a diamond-shaped graph (AR-004).
575     *
576     * @param array<string, true> $seen
577     */
578    private static function collect(Graph $graph, Iri|BlankNode $node, Graph $into, array &$seen): void
579    {
580        $seen[$node->toNTriples()] = true;
581        foreach ($graph->about($node) as $triple) {
582            $into->add(new Triple($triple->subject, $triple->predicate, $triple->object));
583            $object = $triple->object;
584            if (($object instanceof Iri || $object instanceof BlankNode) && !isset($seen[$object->toNTriples()]) && ($object instanceof BlankNode || LogisticsObject::isEmbeddedId($object))) {
585                self::collect($graph, $object, $into, $seen);
586            }
587        }
588    }
589
590    /**
591     * The revision properties are response metadata the server adds to a body
592     * (and the response object carries them); without them the object compares
593     * equal to what the holder stored and can be diffed or republished as is.
594     */
595    private static function withoutRevisionProperties(LogisticsObject $object): LogisticsObject
596    {
597        $graph = new Graph();
598        foreach ($object->graph as $triple) {
599            if ($triple->subject->equals($object->iri) && \in_array($triple->predicate->value, [Api::hasRevision, Api::hasLatestRevision], true)) {
600                continue;
601            }
602            $graph->add($triple);
603        }
604
605        return $object->withGraph($graph);
606    }
607
608    /**
609     * scheme://host:port with the scheme and host lowercased and the default
610     * port spelled out, so "https://A.example" and "https://a.example:443" are
611     * one origin. Null for anything that is not a plain http(s) URL, including
612     * URLs with userinfo, which exist mainly to confuse origin checks.
613     */
614    private static function originOf(string $url): ?string
615    {
616        $parts = parse_url($url);
617        if ($parts === false || !isset($parts['scheme'], $parts['host']) || isset($parts['user'], $parts['pass']) || isset($parts['user'])) {
618            return null;
619        }
620        $scheme = strtolower($parts['scheme']);
621        if (!\in_array($scheme, ['http', 'https'], true)) {
622            return null;
623        }
624        $port = $parts['port'] ?? ($scheme === 'https' ? 443 : 80);
625
626        return $scheme . '://' . strtolower($parts['host']) . ':' . $port;
627    }
628
629    private static function iri(Iri|string $value): Iri
630    {
631        return $value instanceof Iri ? $value : new Iri($value);
632    }
633
634    private static function body(ResponseInterface $response): string
635    {
636        return (string) $response->getBody();
637    }
638
639    private static function location(ResponseInterface $response): Iri
640    {
641        $location = $response->getHeaderLine('Location');
642        if ($location === '') {
643            throw new ClientException(\sprintf('The server answered %d without a Location header.', $response->getStatusCode()));
644        }
645
646        return new Iri($location);
647    }
648
649    private static function httpDate(ResponseInterface $response, string $header): ?DateTimeImmutable
650    {
651        $value = $response->getHeaderLine($header);
652        if ($value === '') {
653            return null;
654        }
655        $parsed = DateTimeImmutable::createFromFormat('D, d M Y H:i:s \G\M\T', $value, new DateTimeZone('UTC'));
656
657        return $parsed === false ? null : $parsed;
658    }
659
660    private static function versionParameter(string $contentType): ?string
661    {
662        return preg_match('/;\s*version\s*=\s*"?([0-9.]+)"?/i', $contentType, $m) === 1 ? $m[1] : null;
663    }
664}