Syncing from an API.

Some systems publish their data through an HTTP endpoint instead of dropping a file. A middleware service in front of an ERP is a common case. It answers GET /api/bookings?updated_since=... with one page of JSON at a time.

An ApiSyncDefinition polls an endpoint like that on a schedule. Each fetch pages through the endpoint and writes every record it returns to one NDJSON file, called a snapshot, in the staging area. The snapshot then goes through the same pipeline as a file feed: the checksum ledger, chunked writes, row failures, archiving, sync:status, and sync:retry.

namespace App\Syncs;

use App\Models\Booking;
use ByRcsc\LaravelDataSync\Definitions\ApiSyncDefinition;
use ByRcsc\LaravelDataSync\Definitions\Field;
use Illuminate\Console\Scheduling\Event;
use Illuminate\Http\Client\PendingRequest;
use Illuminate\Http\Client\Response;

final class BookingsSync extends ApiSyncDefinition
{
    public function endpoint(): string
    {
        return 'https://middleware.example.com/api/bookings';
    }

    public function request(PendingRequest $http, mixed $watermark): PendingRequest
    {
        return $http
            ->withToken(config('services.middleware.token'))
            ->withQueryParameters(['updated_since' => $watermark ?? '1970-01-01']);
    }

    public function nextPage(Response $response): ?string
    {
        return $response->json('meta.next_page_url');
    }

    public function records(Response $response): iterable
    {
        return $response->json('data', []);
    }

    public function watermarkFrom(Response $response, iterable $records): mixed
    {
        return $response->json('meta.max_updated_at');
    }

    public function model(): string
    {
        return Booking::class;
    }

    public function fields(): array
    {
        return [
            Field::make('booking_no', from: 'BookingNo'),
            Field::make('status', from: 'Status'),
        ];
    }

    public function matchOn(): array
    {
        return ['booking_no'];
    }

    public function schedule(Event $event): void
    {
        $event->everyFifteenMinutes();
    }
}

Scaffold one with:

php artisan make:sync BookingsSync --api

Register it in data-sync.syncs like any other definition, then run it with php artisan sync:run bookings.

What changes from a file sync

ApiSyncDefinition extends SyncDefinition. Field mapping, matchOn(), rules(), the write hooks, chunking, atomic(), maxErrors(), payload capture, queue routing, and scheduling work the same way as for a file feed.

Two methods are fixed. format() always returns Format::ndjson(), because the package writes the snapshot itself. source() always points at the definition's directory in the staging area. Both methods are final.

The definition gains these methods:

The package validates these methods when it resolves the definition. endpoint() must return an http or https URL. timeout() must be at least 1, retries() and retryDelay() must not be negative, and lockTimeout() must be at least 1. A failure throws an InvalidDefinitionException that names the method to fix.

Build the request

request() receives the current watermark and a PendingRequest that already carries the timeout and retry settings. Add whatever the endpoint needs and return it:

public function request(PendingRequest $http, mixed $watermark): PendingRequest
{
    return $http
        ->withToken(config('services.middleware.token'))
        ->acceptJson()
        ->withQueryParameters(['updated_since' => $watermark ?? '1970-01-01']);
}

The package calls request() once per fetch and sends every page with the request it returns. The watermark is null until a run first advances it.

The package does not manage credentials. If the endpoint needs a short-lived token, get it here from whatever your application already uses.

Page through the results

The package sends a GET request to endpoint(), then passes each response to nextPage(). Return the absolute URL of the next page, or null on the last page:

public function nextPage(Response $response): ?string
{
    return $response->json('meta.next_page_url');
}

The default returns null, so a definition without nextPage() fetches one page.

A next-page URL keeps its own query string. When the URL and request() set the same parameter, the value in the URL wins. Parameters that only request() sets, such as updated_since, are still sent with every page.

If nextPage() returns a URL that the fetch has already requested, the package stops and fails the fetch. This catches an endpoint that keeps returning the same next-page link on its last page.

Pick the records out of the response

records() returns the records in one response. The default returns the whole decoded JSON body, which suits an endpoint that answers with a bare array. For an endpoint that wraps its rows in an envelope, return the part that holds them:

public function records(Response $response): iterable
{
    return $response->json('data', []);
}

Each record becomes one line of the snapshot, and the NDJSON reader reads it back as one row. A record must be a JSON object. The reader turns every value into a string or null, and a nested object or array arrives as a JSON string. Map fields by the record's keys, for example Field::make('booking_no', from: 'BookingNo').

To store a nested object as JSON, map it to an attribute with a JSON cast, such as array. The package decodes the string before it fills the model, so the column holds the object encoded once. See JSON attributes.

Resume from a watermark

A watermark is the value that tells the remote system where the last fetch stopped, such as an updated_at timestamp or a sequence number. The package keeps one watermark per definition in the sync_watermarks table. It passes the watermark to request() and asks watermarkFrom() for the next one:

public function watermarkFrom(Response $response, iterable $records): mixed
{
    return $response->json('meta.max_updated_at');
}

The package calls watermarkFrom() once per page, with that page's response and records. The last non-null value wins, so return null from a page that has nothing to report, such as an empty last page. Any value that json_encode() accepts can be a watermark.

Prefer a value the remote system reports, as in the example above. A value you compute from $records, such as the highest updated_at on the page, covers the whole fetch only when the endpoint returns records oldest first.

When the watermark moves

The package stores the new value with the snapshot. It moves the watermark only when a run over that snapshot finishes as completed:

When a run does not write every row, the next fetch asks the remote system for the same window again. If the same row fails on every run, the watermark never moves. When the last 5 runs all finish without completed, sync:status and sync:doctor warn about the definition. data-sync.stalled_watermark_runs sets the count. Rows that succeeded the first time are written again, and the matchOn() upsert updates them instead of duplicating them. That depends on a unique index on the match columns, which sync:doctor warns about when it is missing.

A queued run takes time to finish, and the watermark keeps its old value until it does. A fetch that starts in that time asks for the same window, so it can return records that the earlier run is still writing. The same upsert makes that safe.

Each snapshot moves the watermark at most once. After its run completes, reprocessing it with sync:retry or --force leaves the watermark alone. A retry of a run that never completed can still move the watermark, to the value its snapshot carried. If a newer fetch has moved the watermark further since, it moves back, and the next fetch asks for records the package already has. That costs one refetch and loses nothing.

A quiet feed still records a run

A fetch that returns no records still creates a run. The run finishes as completed with zero rows, and sync:status lists it like any other run. A feed with nothing new therefore looks different in the history from a schedule that stopped running.

The ledger never skips an empty snapshot. Every empty snapshot has the same checksum, so skipping them would hide every quiet fetch after the first one and hold the watermark in place.

When a fetch fails

The package retries a request that fails with a connection error, a 5xx status, or a 429 status. It makes retries() extra attempts and waits retryDelay() milliseconds between them. Any other 4xx status fails on the first attempt, because the remote system gives the same answer to the same request.

When a fetch still fails after its retries:

• The package records a failed run and dispatches SyncFailed. The run's error_summary names the endpoint and the HTTP error.
• The pages that arrived before the failure become the snapshot, and the package copies it to the failed area.
• The watermark does not move.
• sync:run prints the error and exits non-zero.

The next scheduled fetch starts from the same watermark, and that is usually the only recovery a failed fetch needs. sync:retry on the failed run processes only the pages that arrived before the failure, and it never moves the watermark.

The snapshot is the audit record

The package names each snapshot after the definition and the fetch time, for example bookings-20261109-141500.ndjson. The snapshot follows the same rules as a staged file:

• The package records its checksum in the sync_files ledger. A fetch that returns the same bytes as a completed snapshot is skipped unless you pass --force.
• A snapshot whose run finished as completed_with_errors does not count as completed. If the next fetch returns the same bytes, the package processes them again.
• A completed run archives the snapshot. A failed or cancelled run copies it to the failed area. sync:prune removes both copies on the usual retention windows.
• sync:retry restores the snapshot, so a retry processes the records the API returned at the time of the fetch, not what it returns now.
• Like a staged file, the copy in the staging area stays after a successful run. See what sync:prune does not clean.

Run modes

sync:run bookings fetches in the command's process, then dispatches the snapshot to the queue in the same way as a discovered file. The scheduler does the same. With --now, the command also processes the snapshot in its own process.

--dry-run fetches from the endpoint and writes a snapshot, then counts rows without writing them. It does not mark the checksum as processed and does not move the watermark. The remote system still receives the requests, so a dry run is safe only against an endpoint where a GET changes nothing.

--force processes the snapshot even when the ledger has a completed copy of the same bytes. It still sends the current watermark to request(). To fetch from an earlier point, change the watermark row in sync_watermarks.

A fetch takes the same per-definition cache lock as file discovery. A scheduled fetch that starts while another fetch of the same definition is running reports Skipped [bookings] because discovery is already running. and exits zero. The lock expires after lockTimeout() seconds, 300 by default. Set it above the longest fetch you expect, or a second fetch can start while the first is still paging.

Send acknowledgements back

The package only pulls. Your application sends anything back to the remote system, such as a status update for each booking it received. Use a listener on SyncCompleted for that:

use ByRcsc\LaravelDataSync\Enums\RunStatus;
use ByRcsc\LaravelDataSync\Events\SyncCompleted;

public function handle(SyncCompleted $event): void
{
    if ($event->run->definition !== 'bookings'
        || $event->run->dry_run
        || $event->run->status !== RunStatus::Completed) {
        return;
    }

    AcknowledgeBookings::dispatch($event->run->id);
}

SyncCompleted fires for completed_with_errors runs too, so check the status when you only acknowledge complete windows. Queue the listener. The package dispatches its events synchronously.

afterSave() can also call the remote system, once for each row:

public function afterSave(Model $model, array $row): void
{
    Http::withToken(config('services.middleware.token'))
        ->patch("https://middleware.example.com/api/bookings/{$model->booking_no}", [
            'status' => 'received',
        ]);
}

Declaring afterSave() switches the definition to the Eloquent write strategy, and the hook runs inside the chunk's transaction. A slow call holds that transaction open. A chunk that rolls back after the call has already acknowledged rows that were never saved. Prefer the listener unless the remote system needs one call per row as it lands.

Either way, these calls are your application's own HTTP traffic. The package does not record them in sync_runs. A queued listener that fails does not fail the run that triggered it.

Limits

• Polling only. The package fetches on a schedule. Nothing listens for a remote system to call your application, so webhooks and inbound push are out of scope.
• GET requests only. An endpoint that needs a POST body to return records cannot be a source.
• No token refresh. request() receives a PendingRequest, and your application applies credentials it already manages.
• The fetch finishes before processing starts. The package holds one page in memory at a time and spills the snapshot to a temporary file as it grows. Processing begins only after the last page arrives.
• No writes back. Two-way sync is out of scope. See send acknowledgements back for the pattern to use instead.

What to read next

• Field mapping for mapping record keys to model attributes.
• Scheduling for how often the fetch runs.
• Testing for faking the endpoint in a test.
• Syncing from a database when the records sit in another database instead.