One interface, eight job sources: designing a connector pipeline that doesn't fight you

I built DevRadar as a personal dashboard to track developer job offers across the French market. Nothing fancy — a Laravel app, a few dozen offers a day, one user (me). But it pulls from eight very different sources: a government API with OAuth2, two HTML pages I scrape by hand, a couple of official REST APIs, an RSS feed, and — most recently — a mailbox full of job-alert emails.

Those sources have almost nothing in common at the transport level. What they do have in common is what happens after the data arrives: turn it into a job offer, figure out if we've seen it before, save it, keep a record of the sync. That's the part worth designing well, because it's the part every future source has to go through.

This post walks through how that pipeline is built — two files, SynchronizeSourceAction and PersistOfferAction, plus the small set of types that connect them — and why the design paid for itself the day I added a source that isn't even a network request.

The shape of the problem

Here's the connector list as it stands in docs/job-sources.md:

Every one of these needs different auth, different pagination, different error modes, different parsing. A malformed HTML page, an empty RSS feed, and an IMAP connection timeout are not remotely the same failure — but by the time the data reaches persistence, none of that should matter anymore.

So the pipeline draws one hard line: a connector's only job is to produce a list of NormalizedOffer DTOs. Everything before that line is source-specific and lives in app/Sync/Connectors/. Everything after it is source-agnostic and lives in app/Actions/Sync/.

The contract

interface Connector
{
    public function driver(): string;

    /** @throws InvalidConnectorConfigurationException */
    public function validateConfiguration(Source $source): void;

    /** @throws ConnectorFetchException */
    public function fetch(Source $source, ?string $cursor = null): ConnectorPage;
}

That's the whole interface. fetch() returns a ConnectorPage:

final class ConnectorPage
{
    public function __construct(
        public readonly array $offers,       // list<NormalizedOffer>
        public readonly ?string $nextCursor = null,
        public readonly bool $hasMorePages = false,
    ) {}
}

A source with no real pagination (most of the HTML ones) just always returns hasMorePages: false. A source with cursor-based pagination (the APIs) hands back whatever token means "continue from here" — a page number, an offset, an API-specific cursor string. The orchestrator doesn't care which; it just keeps asking until hasMorePages is false.

And every offer that comes out the other end, regardless of source, is the same shape:

final class NormalizedOffer
{
    public function __construct(
        public readonly ?string $externalId = null,
        public readonly ?string $title = null,
        public readonly ?string $company = null,
        public readonly ?string $location = null,
        public readonly ?string $contractType = null,   // raw string — mapped later
        public readonly ?string $remoteStatus = null,    // raw string — mapped later
        public readonly ?int $salaryMinimum = null,
        public readonly array $technologies = [],
        public readonly ?CarbonImmutable $publishedAt = null,
        public readonly array $rawPayload = [],
        public readonly array $evidence = [],
        public readonly ?float $extractionConfidence = null,
        // ...
    ) {}
}

Two things worth calling out, because they're the two rules that keep eight connectors from turning into eight slightly-different dialects:

Every field is nullable, and a connector must leave it null when the source doesn't provide it — never guess. If Jooble's alert email doesn't give me a job description, description stays null. It doesn't get padded with the title, it doesn't get inferred from the company name. An empty field is honest data; a guessed field is a bug waiting to surface three steps downstream.

Contract type and remote policy are passed as raw strings, not enums. A connector doesn't decide whether "CDI" or "temps plein" means ContractType::Cdi — it just hands over whatever text the source gave it, and a shared mapper does the classification:

final class OfferFieldMapper
{
    public static function mapContractType(?string $raw): ?ContractType
    {
        $needle = self::normalize($raw); // fold accents, lowercase, collapse whitespace

        return match (true) {
            str_contains($needle, 'cdi'), str_contains($needle, 'temps plein') => ContractType::Cdi,
            str_contains($needle, 'freelance'), str_contains($needle, 'independant') => ContractType::Freelance,
            str_contains($needle, 'alternance'), str_contains($needle, 'apprentissage') => ContractType::Alternance,
            // ...
            default => null,
        };
    }
}

This one function is why an HTML connector scraping French badges ("✅ CDI") and a REST API returning employmentType: FULL_TIME both end up with the same ContractType::Cdi — one vocabulary, one place it's defined, zero connectors reimplementing it.

The orchestrator: SynchronizeSourceAction

This is the part that runs a sync from start to finish, and it's deliberately boring:

final class SynchronizeSourceAction
{
    public function execute(Source $source): SyncRun
    {
        $lock = Cache::lock("sync:source:{$source->id}", 600);
        if (! $lock->get()) {
            throw new SourceSyncInProgressException(/* ... */);
        }

        try {
            return $this->synchronize($source);
        } finally {
            $lock->release();
        }
    }

    private function synchronize(Source $source): SyncRun
    {
        $connector = $this->registry->resolve($source);
        $connector->validateConfiguration($source);

        $cursor = null;
        $seenOfferIds = [];

        do {
            $page = $connector->fetch($source, $cursor);

            foreach ($page->offers as $dto) {
                $persisted = $this->persistOffer->execute($source, $dto);
                $seenOfferIds[] = $persisted->offerId;
            }

            $cursor = $page->nextCursor;
        } while ($page->hasMorePages && $cursor !== null);

        $this->expireStaleOffers($source, $seenOfferIds);
        // ...record a SyncRun with stats, return it
    }
}

Three decisions here are doing more work than the line count suggests:

A per-source lock, not a global one. Two syncs of the same source can't overlap (no point double-fetching the same French Travail search), but syncing eight sources in parallel is completely fine, because they never touch each other's lock key.

The Connector is resolved from a registry keyed by a driver string stored on the Source row, not hardcoded per source. Adding source #9 means writing a class and adding one line to config/connectors.php — nothing about SynchronizeSourceAction changes, ever.

Stale-offer expiry is a side effect of "what did we see this run," not a separate job. Every offer ID that came back from persistOffer->execute() goes into $seenOfferIds. After the loop, anything this source owns that's still marked New but wasn't in that set gets flipped to Expired. No offer is ever deleted — a posting that's gone from the source just stops being active, and if it comes back next run, PersistOfferAction un-expires it automatically. This assumes a connector returns its complete current listing every run, which is true for all eight sources today; a delta-only API would need a different reconciliation strategy, and that's written down as a known limitation rather than quietly assumed away.

The interesting part: PersistOfferAction

Deduplication across eight independently-scraped sources is the part that could have turned into a mess, so it's worth showing the actual decision tree. OfferDeduplicator::findMatch() checks four signals, in order of how much they can be trusted:

// Level 1 — same source, same external id → definitely the same offer
$offer = Offer::where('source_id', $source->id)->where('external_id', $externalId)->first();
if ($offer) return new OfferMatch($offer, strong: true);

// Level 2 — identical canonical URL, any source → definitely the same offer
$offer = Offer::where('canonical_url', $canonicalUrl)->first();
if ($offer) return new OfferMatch($offer, strong: true);

// Level 3 — same content fingerprint (title+company+description hash) → *probably*
$offer = Offer::where('content_hash', $contentHash)->first();
if ($offer) return new OfferMatch($offer, strong: false);

// Level 4 — same title+company(+location) → *probably*
// ...

The strong flag is the whole trick. Levels 1 and 2 are exact evidence — the same job board can't hand you two different external IDs for one posting, and two sources publishing the identical canonical URL are, by construction, the same listing. Those get merged: the existing row is updated in place.

Levels 3 and 4 are signals, not proof. A title/company/location match could easily be two different postings for the same role at the same company, published a week apart. So a non-strong match never merges automatically — it creates a new offer row and tags it possible_duplicate_of for manual review. I'd rather show a false-positive "possible duplicate" once in a while than silently collapse two distinct job postings into one and lose data.

From there, PersistOfferAction branches into exactly three outcomes:

• No match → create. Straightforward insert, technologies attached, done.
• Strong match, same source owns it → update. Title, salary, description, everything gets refreshed. If the offer had expired and reappeared in the feed, its status flips back to New — but a manual decision (favorited, archived, ignored) is never touched by an automated sync.
• Strong match, a different source owns it → enrich, don't overwrite. This is the one that surprised me the most when I first wrote it. If Adzuna and Free-Work both list the same posting and Free-Work saw it first, Adzuna's copy doesn't overwrite Free-Work's data — it only fills in fields Free-Work left null. The owning source's data always wins; a later source can only patch gaps, never clobber.

private function enrichForeignOffer(Offer $offer, NormalizedOffer $dto): void
{
    $gaps = array_filter([
        'company' => $dto->company,
        'salary_min' => $dto->salaryMinimum,
        'location' => $dto->location,
        // ...
    ], fn ($value) => $value !== null);

    foreach ($gaps as $attribute => $value) {
        if ($offer->{$attribute} === null) {
            $offer->{$attribute} = $value;
        }
    }

    $offer->save();
}

Where it got tested for real: adding a source that isn't a network request

Every connector I described above does the same thing at the transport level: send an HTTP request, get a response, parse it. Then I wanted to add Jooble — a French job aggregator — as a source. The catch: their site sits behind a Cloudflare bot challenge on every page (including robots.txt-adjacent ones), and their terms of service explicitly prohibit automated crawlers. Their official API exists but is out of scope for how I wanted to use it here.

What Jooble does do is email me job alerts, because I subscribed to one like anyone would. So instead of an HTTP connector, I wrote one that reads a mailbox over IMAP:

final class JoobleEmailConnector implements Connector
{
    public function __construct(
        private readonly MailboxContract $mailbox,
        private readonly JoobleEmailParser $parser,
    ) {}

    public function fetch(Source $source, ?string $cursor = null): ConnectorPage
    {
        $messages = $this->mailbox->fetchUnseenMessages(
            config('connectors.jooble_email.max_messages_per_sync'),
            JoobleEmailParser::SENDER_DOMAIN,
        );

        $offers = [];
        foreach ($messages as $message) {
            if (! $this->parser->supports($message)) {
                continue; // not a Jooble email — leave it unread for another parser
            }

            $offers = [...$offers, ...$this->parser->parse($message)];
            $this->mailbox->markProcessed($message);
        }

        return new ConnectorPage(offers: $offers, nextCursor: null, hasMorePages: false);
    }
}

This is still just a Connector. SynchronizeSourceAction doesn't know or care that fetch() opened an IMAP session instead of an HTTP one. PersistOfferAction doesn't know or care that the NormalizedOffers it's deduplicating came from parsing <span class="tag"> badges in an HTML email instead of a JSON-LD block. The entire sync pipeline — locking, pagination, stale-offer expiry, the four-level dedup, junior/senior classification — is exactly as much code as it was before this source existed: zero new lines.

The IMAP "unread flag" turned out to be a nice fit for the cursor concept too — a message only gets flagged \Seen once it's been successfully parsed, so "what's new since last time" falls out of the mailbox itself instead of needing a persisted cursor. (I did get this wrong on the first pass: fetching "the first N unread messages" without filtering by sender first meant a mailbox with 30,000 unread emails buried the ~450 Jooble ones completely. The fix was filtering server-side, over IMAP, before applying the limit — a good reminder that "unread" and "relevant" are not the same predicate, and testing against realistic mailbox contents rather than a couple of fixtures would have caught it sooner.)

What I'd tell someone building this from scratch

• Draw the abstraction boundary at the data, not the transport. "Fetch and parse" is where sources genuinely differ; "decide if this is a new offer" is where they don't. Put the interface exactly on that seam.
• Nullable-by-default DTOs make "we don't know" a real, representable state, instead of a temptation to guess. It costs you some ?string noise; it saves you from a parser inventing data three months later when nobody remembers it did.
• Not all duplicate signals deserve the same trust, and conflating them is how you either merge two different jobs into one or spam yourself with false duplicates. Give strong evidence the power to merge and weak evidence only the power to flag.
• The real test of an abstraction is adding something that doesn't fit the mold you designed it around. For me that was a mailbox instead of an HTTP client. If your interface survives that with zero changes to the orchestrator, it was drawn in the right place.