successful()) { return NodeRunResult::failed('Feed request failed.', ['status' => $response->status()]); } try { $xml = new SimpleXMLElement($response->body()); } catch (Throwable $e) { Log::warning('Fetch RSS node: malformed feed', [ 'run_id' => $run->id, 'feed_url' => $feedUrl, 'error' => $e->getMessage(), ]); return NodeRunResult::failed('Malformed RSS feed.'); } $nodeId = (string) $run->current_node_id; // Dry runs bypass the watermark entirely: they neither load nor advance // the persisted state, AND they use an epoch watermark so every item in // the feed is treated as new — otherwise a "now" default would silently // return zero items on any feed that hasn't published in the last second. $state = $run->is_dry_run ? null : AutomationNodeState::for($run->automation_id, $nodeId); $watermark = $run->is_dry_run ? CarbonImmutable::createFromTimestamp(0) : $this->parseWatermark(data_get($state->data, 'last_item_date')); [$newItems, $newestSeen] = $this->collectNewItems($xml, $watermark); if ($state !== null && $newestSeen !== null) { $state->update(['data' => array_merge($state->data ?? [], [ 'last_item_date' => $newestSeen->toIso8601String(), ])]); } if ($newItems === []) { return NodeRunResult::completed(['fetch' => ['count' => 0]], nextHandle: 'no_items'); } $first = array_shift($newItems); if (! $run->is_dry_run) { $this->spawnSiblings($run, $nodeId, $newItems); } return NodeRunResult::completed([ 'fetch' => ['count' => count($newItems) + 1, 'spawned' => $run->is_dry_run ? 0 : count($newItems)], 'fetched' => $first, ]); } private function collectNewItems(SimpleXMLElement $xml, CarbonImmutable $watermark): array { $items = []; $newestSeen = null; foreach ($xml->channel->item ?? [] as $item) { $key = (string) ($item->guid ?? $item->link); if ($key === '') { continue; } $pubDate = $this->parsePubDate((string) $item->pubDate); if ($pubDate === null) { continue; } if ($newestSeen === null || $pubDate->greaterThan($newestSeen)) { $newestSeen = $pubDate; } if (! $pubDate->greaterThan($watermark)) { continue; } $items[] = [ '_pubDate' => $pubDate, 'key' => $key, 'title' => (string) $item->title, 'link' => (string) $item->link, 'description' => (string) $item->description, 'pubDate' => (string) $item->pubDate, ]; } // Process oldest-first so siblings inherit a stable order matching feed chronology. usort($items, fn ($a, $b) => $a['_pubDate']->getTimestamp() <=> $b['_pubDate']->getTimestamp()); // Drop the internal sort key — downstream nodes shouldn't see it. $items = array_map(function (array $item): array { unset($item['_pubDate']); return $item; }, $items); return [$items, $newestSeen]; } private function spawnSiblings(AutomationRun $parent, string $fetchNodeId, array $items): void { if ($items === []) { return; } $nextNodeId = $this->findNextNodeId($parent, $fetchNodeId); foreach ($items as $item) { $sibling = AutomationRun::create([ 'automation_id' => $parent->automation_id, 'is_manual' => $parent->is_manual, 'is_dry_run' => $parent->is_dry_run, 'status' => RunStatus::Pending, 'context' => array_merge($parent->context ?? [], ['fetched' => $item]), ]); if ($nextNodeId === null) { $sibling->update(['status' => RunStatus::Completed, 'finished_at' => now()]); continue; } ProcessAutomationNode::dispatch($sibling, $nextNodeId); } } private function findNextNodeId(AutomationRun $run, string $fromNodeId): ?string { $connection = collect($run->automation->connections ?? []) ->first(fn ($c) => $c['source'] === $fromNodeId && ($c['source_handle'] ?? 'default') === 'default'); return $connection['target'] ?? null; } private function parseWatermark(?string $stored): CarbonImmutable { if ($stored === null) { return CarbonImmutable::now(); } try { return CarbonImmutable::parse($stored); } catch (Throwable) { return CarbonImmutable::now(); } } private function parsePubDate(string $raw): ?CarbonImmutable { if ($raw === '') { return null; } try { return CarbonImmutable::parse($raw); } catch (Throwable) { return null; } } }